nettyConfig.addProperty("child.tcpNoDelay", "true");
nettyConfig.addProperty("child.keepAlive", "true");
final AsyncHttpClient c = new AsyncHttpClient(b.setAsyncHttpClientProviderConfig(nettyConfig).build());
Client client = ClientFactory.getDefault().newClient();
RequestBuilder request = client.newRequestBuilder();
request.method(Request.METHOD.GET).uri(url);
request.transport(Request.TRANSPORT.WEBSOCKET);
request.header("X-wakeUpNIO", "true");
final CountDownLatch l = new CountDownLatch(clientNum);
final CountDownLatch messages = new CountDownLatch(messageNum * clientNum);
long clientCount = l.getCount();
final AtomicLong total = new AtomicLong(0);
Socket[] sockets = new Socket[clientNum];
for (int i = 0; i < clientCount; i++) {
final AtomicLong start = new AtomicLong(0);
sockets[i] = client.create(client.newOptionsBuilder().runtime(c).reconnect(false).build())
.on(new Function<Integer>() {
@Override
public void on(Integer statusCode) {
start.set(System.currentTimeMillis());
l.countDown();
}
}).on(new Function<String>() {
int mCount = 0;
@Override
public void on(String s) {
if (s.startsWith("message")) {
String[] m = s.split("\n\r");
mCount += m.length;
messages.countDown();
System.out.println("Message left receive " + messages.getCount() + " message " + s);
if (mCount == messageNum) {
// System.out.println("All messages received " + mCount);
total.addAndGet(System.currentTimeMillis() - start.get());
}
}
}
}).on(new Function<Throwable>() {
@Override
public void on(Throwable t) {
t.printStackTrace();
}
});
}
for (int i = 0; i < clientCount; i++) {
sockets[i].open(request.build());
}
l.await(30, TimeUnit.SECONDS);
System.out.println("OK, all Connected: " + clientNum);
Socket socket = client.create(client.newOptionsBuilder().runtime(c).build());
socket.open(request.build());
for (int i = 0; i < messageNum; i++) {
socket.fire("message" + i);
}
messages.await(1, TimeUnit.HOURS);
socket.close();