package org.jgroups.tests;

import org.jgroups.Global;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.protocols.TP;
import org.jgroups.util.Util;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author Bela Ban
 * @version $Id: TransportThreadPoolTest.java,v 1.8 2008/08/08 17:07:11 vlada Exp $
 */
@Test(groups=Global.STACK_DEPENDENT,sequential=true)
public class TransportThreadPoolTest extends ChannelTestBase {
    JChannel c1, c2;

    @BeforeMethod
    protected void setUp() throws Exception {
        c1=createChannel(true, 2);
        c2=createChannel(c1);
    }

    @AfterMethod
    protected void tearDown() throws Exception {
        Util.close(c2, c1);
    }


    @Test
    public void testThreadPoolReplacement() throws Exception {
        Receiver r1=new Receiver(), r2=new Receiver();
        c1.setReceiver(r1);
        c2.setReceiver(r2);
        c1.connect("TransportThreadPoolTest");
        c2.connect("TransportThreadPoolTest");

        c1.send(null, null, "hello world");
        c2.send(null, null, "bela");

        Util.sleep(500); // need to sleep because message sending is asynchronous
        assert r1.getMsgs().size() == 2;
        assert r2.getMsgs().size() == 2;

        TP transport=c1.getProtocolStack().getTransport();
        ExecutorService thread_pool=Executors.newCachedThreadPool();
        transport.setDefaultThreadPool(thread_pool);

        transport=c2.getProtocolStack().getTransport();
        thread_pool=Executors.newCachedThreadPool();
        transport.setDefaultThreadPool(thread_pool);

        c1.send(null, null, "message 3");
        c2.send(null, null, "message 4");
        Util.sleep(500);

        System.out.println("messages c1: " + print(r1.getMsgs()) + "\nmessages c2: " + print(r2.getMsgs()));
        assert r1.getMsgs().size() == 4;
        assert r2.getMsgs().size() == 4;
    }


    private static String print(List<Message> msgs) {
        StringBuilder sb=new StringBuilder();
        for(Message msg: msgs) {
            sb.append("\"" + msg.getObject() + "\"").append(" ");
        }
        return sb.toString();
    }


    private static class Receiver extends ReceiverAdapter {
        List<Message> msgs=new LinkedList<Message>();

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

        public void receive(Message msg) {
            msgs.add(msg);
        }
    }
}

@@ -376,24 +376,68 @@
 .util.List;%0A
+import java.util.concurrent.CountDownLatch;%0A
 import java.
@@ -507,16 +507,54 @@
 ecutors;
+%0Aimport java.util.concurrent.TimeUnit;
 %0A%0A/**%0A *
@@ -625,29 +625,29 @@
 v 1.
-8
+9
  200
-8/08/08 17:07:11
+9/06/04 07:29:49
  vla
@@ -1400,173 +1400,8 @@
 );%0A%0A
-        Util.sleep(500); // need to sleep because message sending is asynchronous%0A        assert r1.getMsgs().size() == 2;%0A        assert r2.getMsgs().size() == 2;%0A%0A
     
@@ -1837,22 +1837,122 @@
     
-Util.sleep(500
+%0A        r1.getLatch().await(2000, TimeUnit.MILLISECONDS);%0A        r1.getLatch().await(2000, TimeUnit.MILLISECONDS
 );%0A%0A
@@ -2510,16 +2510,85 @@
 sage%3E();
+%0A        %0A        final CountDownLatch latch = new CountDownLatch(4);
 %0A%0A      
@@ -2652,24 +2652,110 @@
 s;%0A        %7D
+%0A        %0A        public CountDownLatch getLatch()%7B%0A           return latch;%0A        %7D
 %0A%0A        pu
@@ -2814,16 +2814,47 @@
 d(msg);%0A
+            latch.countDown();%0A
         

use CountDownLatch instead of Util.sleep