package org.jgroups.tests;

import org.jgroups.*;
import org.testng.annotations.Test;

import java.io.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.List;
import java.util.ArrayList;

/**
 * 
 * coord.stopFlush() and coord.close() called in sequence cause flush lockup
 * See https://jira.jboss.org/jira/browse/JGRP-959
 * 
 * 
 * @author vladimir
 * @since 2.8
 */
@Test(groups = Global.FLUSH, sequential = true)
public class FlushCloseOpenTest extends ChannelTestBase {

   @Test
   public void testAndLoop() throws Exception {

      for (int i = 1; i <= 4; i++) {
         Channel channel = createChannel(true, 2);
         ReceiverImpl receiver = new ReceiverImpl();
         channel.setReceiver(receiver);
         channel.connect("testClust");

         Channel channel2 = createChannel((JChannel) channel);
         ReceiverImpl receiver2 = new ReceiverImpl();
         channel2.setReceiver(receiver2);
         channel2.connect("testClust");

         sendMessage(channel, "msg1");
         sendMessage(channel2, "msg2");

         channel.startFlush(false);
         assertCount(receiver, 2, receiver2, 2);
         channel.stopFlush();

         channel.close();
         channel = createChannel((JChannel) channel2);
         channel.setReceiver(receiver);
         channel.connect("testClust");
         sendMessage(channel2, "msg3");

         channel2.startFlush(false);
         assertCount(receiver, 3, receiver2, 3);
         channel.stopFlush();
         channel2.close();
         channel2 = createChannel((JChannel) channel);
         channel2.setReceiver(receiver2);
         channel2.connect("testClust");

         sendMessage(channel2, "msg4");
         channel2.startFlush(true);
         assertCount(receiver, 4, receiver2, 4);

         channel.close();
         channel2.close();
         receiver.receiveCount.set(0);
         receiver2.receiveCount.set(0);
         System.out.println("***** Round " + i + " done *****");
      }
   }

   private  void sendMessage(Channel channel, Serializable obj) throws Exception {
      if (!channel.isConnected()) {
         log.warn("Channel disconnected in send, discarding msg");
         return;
      }
      Message msg = new Message(null, null, obj);
      log.debug("Sending message: " + msg);
      channel.send(msg);
      log.debug("Sent message: " + msg);
   }

   private void assertCount(ReceiverImpl srv1, long srv1Count, ReceiverImpl srv2, long srv2Count)
            throws InterruptedException {
      long start = System.currentTimeMillis();
      for (int i = 0; i < 1000; i++) {
         if (srv1.receiveCount.get() == srv1Count && srv2.receiveCount.get() == srv2Count) {
            break;
         }
         Thread.sleep(10L);
      }
      assert srv1Count == srv1.receiveCount.get() : "expected " + srv1Count + " but got "
               + srv1.receiveCount;
      assert srv2Count == srv2.receiveCount.get() : "expected " + srv2Count + " but got "
               + srv2.receiveCount;
      log.info("assert OK in " + (System.currentTimeMillis() - start) + "ms");
   }

   private class ReceiverImpl extends ReceiverAdapter {
      final List<Object> msgs = new ArrayList<Object>();
      public final AtomicLong receiveCount = new AtomicLong();

      public List<Object> getMsgs() {
         return msgs;
      }

      @Override
      public void receive(Message msg) {

         try {
            Object data = msg.getObject();
            msgs.add(data);
            receiveCount.incrementAndGet();
            log.debug("Received msg: " + data);
         } catch (Exception e) {
            log.error("Receive failed", e);
         }
      }

      @Override
      public void viewAccepted(View new_view) {
      }
   }
}
@@ -39,24 +39,54 @@
 .jgroups.*;%0A
+import org.jgroups.util.Util;%0A
 import org.t
@@ -109,16 +109,16 @@
 s.Test;%0A
-
 %0Aimport 
@@ -747,32 +747,63 @@
 iver(receiver);%0A
+         channel.setName(%22A%22);%0A
          channel
@@ -977,32 +977,63 @@
 ver(receiver2);%0A
+         channel.setName(%22B%22);%0A
          channel
@@ -1142,30 +1142,34 @@
 ;%0A%0A         
-channe
+assert Uti
 l.startFlush
@@ -1161,37 +1161,39 @@
 Util.startFlush(
-false
+channel
 );%0A         asse
@@ -1372,32 +1372,63 @@
 iver(receiver);%0A
+         channel.setName(%22A%22);%0A
          channel
@@ -1446,24 +1446,25 @@
 estClust%22);%0A
+%0A
          sen
@@ -1497,32 +1497,35 @@
 ;%0A%0A         
-channel2
+assert Util
 .startFlush(
@@ -1528,13 +1528,16 @@
 ush(
-false
+channel2
 );%0A 
@@ -1730,32 +1730,63 @@
 ver(receiver2);%0A
+         channel.setName(%22B%22);%0A
          channel
@@ -1859,24 +1859,27 @@
         
-channel2
+assert Util
 .startFl
@@ -1882,20 +1882,24 @@
 rtFlush(
-true
+channel2
 );%0A     
@@ -1942,16 +1942,56 @@
 r2, 4);%0A
+         channel2.stopFlush();%0A         
 %0A       

use Util.startFlush and give meaningful names to channels