/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.apache.geode.redis;

import static org.assertj.core.api.Assertions.assertThat;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;

import org.junit.After;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

import org.apache.geode.redis.mocks.MockSubscriber;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.rules.RedisPortSupplier;
import org.apache.geode.test.junit.rules.ExecutorServiceRule;

public abstract class AbstractCommandPipeliningIntegrationTest implements RedisPortSupplier {
  private Jedis publisher;
  private Jedis subscriber;
  private static final int REDIS_CLIENT_TIMEOUT =
      Math.toIntExact(GeodeAwaitility.getTimeout().toMillis());

  @ClassRule
  public static ExecutorServiceRule executor = new ExecutorServiceRule();

  @Before
  public void setUp() {
    subscriber = new Jedis("localhost", getPort(), REDIS_CLIENT_TIMEOUT);
    publisher = new Jedis("localhost", getPort(), REDIS_CLIENT_TIMEOUT);
  }

  @After
  public void tearDown() {
    subscriber.close();
    publisher.close();
  }

  @Test
  public void whenPipelining_commandResponsesAreNotCorrupted() {
    List<String> expectedMessages = Arrays.asList("hello");

    MockSubscriber mockSubscriber = new MockSubscriber();

    Runnable runnable = () -> {
      subscriber.subscribe(mockSubscriber, "salutations");
    };

    Thread subscriberThread = new Thread(runnable);
    subscriberThread.start();
    waitFor(() -> mockSubscriber.getSubscribedChannels() == 1);

    publisher.sadd("foo", "bar");

    // Publish and smembers in a pipeline
    Pipeline pipe = publisher.pipelined();

    pipe.publish("salutations", "hello");
    pipe.smembers("foo");

    List<Object> responses = pipe.syncAndReturnAll();

    mockSubscriber.unsubscribe("salutations");
    waitFor(() -> mockSubscriber.getSubscribedChannels() == 0);
    waitFor(() -> !subscriberThread.isAlive());

    assertThat(mockSubscriber.getReceivedMessages()).isEqualTo(expectedMessages);
  }

  private void waitFor(Callable<Boolean> booleanCallable) {
    GeodeAwaitility.await()
        .ignoreExceptions() // ignoring socket closed exceptions
        .until(booleanCallable);
  }
}

@@ -2924,16 +2924,973 @@
 );%0A  %7D%0A%0A
+  @Test%0A  public void should_returnResultsOfPipelinedCommands_inCorrectOrder() %7B%0A    Jedis jedis = new Jedis(%22localhost%22, getPort(), REDIS_CLIENT_TIMEOUT);%0A    final int NUMBER_OF_COMMANDS_IN_PIPELINE = 100;%0A    int numberOfPipeLineRequests = 1000;%0A%0A    do %7B%0A      Pipeline p = jedis.pipelined();%0A      for (int i = 0; i %3C NUMBER_OF_COMMANDS_IN_PIPELINE; i++) %7B%0A        p.echo(String.valueOf(i));%0A      %7D%0A%0A      List%3CObject%3E results = p.syncAndReturnAll();%0A%0A      verifyResultOrder(NUMBER_OF_COMMANDS_IN_PIPELINE, results);%0A      numberOfPipeLineRequests--;%0A    %7D while (numberOfPipeLineRequests %3E 0);%0A%0A    jedis.flushAll();%0A    jedis.close();%0A  %7D%0A%0A  private void verifyResultOrder(final int numberOfCommandInPipeline, List%3CObject%3E results) %7B%0A    for (int i = 0; i %3C numberOfCommandInPipeline; i++) %7B%0A      String expected = String.valueOf(i);%0A      String currentVal = (String) results.get(i);%0A%0A      assertThat(currentVal).isEqualTo(expected);%0A    %7D%0A  %7D%0A%0A
   privat

Create test to validate ordering of redis pipeline commands (#5552)