PubSub { public static void publish(String topic, Message message) { client().publish(topic, message); }  static void main(String[] args); static PubSubClient startClient(String serverHostName, int serverPort, int maxPendingMessages, boolean isDaemon); static PubSubServer startServer(); static PubSubServer startServer(String bindToHostName, int bindToPort); static PubSubClient client(); static void join(); static synchronized void close(); static void subscribe(String topic, Subscriber<?> subscriber); static void unsubscribe(String topic, Subscriber<?> subscriber); static void publish(String topic, Message message); static void publish(String topic, String message); static boolean awaitFlush(long timeout);  }
@Test public void testWaitUntilEmpty() throws IOException, InterruptedException { PubSubClient client = new PubSubClient(host, port, 0); client.publish("hi", PubSubProtos.StringMessage.newBuilder().setMessage("hello").build()); client.publish("hi", PubSubProtos.StringMessage.newBuilder().setMessage("hello").build()); client.publish("hi", PubSubProtos.StringMessage.newBuilder().setMessage("hello").build()); assertEquals(3, client.pending.size()); assertEquals(false, client.waitUntilEmpty(100)); assertEquals(false, client.waitUntilEmpty(100)); client.start(); assertEquals(true, client.waitUntilEmpty(100)); assertEquals(0, client.pending.size()); client.suspend(); assertEquals(0, client.pending.size()); client.publish("hi", PubSubProtos.StringMessage.newBuilder().setMessage("hello").build()); client.publish("hi", PubSubProtos.StringMessage.newBuilder().setMessage("hello").build()); assertEquals(2, client.pending.size()); client.resume(); assertEquals(true, client.waitUntilEmpty(100)); assertEquals(0, client.pending.size()); client.suspend(); assertEquals(0, client.pending.size()); client.publish("hi", PubSubProtos.StringMessage.newBuilder().setMessage("hello").build()); client.publish("hi", PubSubProtos.StringMessage.newBuilder().setMessage("hello").build()); assertEquals(2, client.pending.size()); assertEquals(false, client.waitUntilEmpty(100)); client.resume(); assertEquals(true, client.waitUntilEmpty(100)); assertEquals(0, client.pending.size()); }