final IConnection client = context.connect(null, "localhost", port);
lock.lock();
List<TaskMessage> list = new ArrayList<TaskMessage>();
TaskMessage message = new TaskMessage(task, req_msg.getBytes());
list.add(message);
client.send(message);
System.out.println("Send first");
JStormUtils.sleepMs(10000);
System.out.println("Begin to Send second");
client.send(message);
System.out.println("Send second");
JStormUtils.sleepMs(15000);
client.send(message);
System.out.println("Send third time");
try {
clientClose.await();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
client.close();
contextClose.signal();
lock.unlock();
}
}).start();
server = context.bind(null, port);
WaitStrategy waitStrategy = (WaitStrategy) Utils
.newInstance((String) storm_conf
.get(Config.TOPOLOGY_DISRUPTOR_WAIT_STRATEGY));
DisruptorQueue recvQueue = new DisruptorQueue(
"NettyUnitTest", ProducerType.SINGLE, 1024, waitStrategy);
server.registerQueue(recvQueue);
TaskMessage recv = server.recv(0);
System.out.println("Receive first");
Assert.assertEquals(req_msg, new String(recv.message()));
server.close();
System.out.println("!!shutdow server and sleep 30s, please wait!!");
Thread.sleep(30000);
IConnection server2 = context.bind(null, port);
server2.registerQueue(recvQueue);
System.out.println("!!!!!!!!!!!!!!!!!!!! restart server !!!!!!!!!!!");
TaskMessage recv2 = server2.recv(0);
Assert.assertEquals(req_msg, new String(recv2.message()));
lock.lock();
clientClose.signal();
server2.close();
contextClose.await();