package com.inmobi.messaging.publisher.examples;

import java.nio.ByteBuffer;

import com.inmobi.messaging.Message;
import com.inmobi.messaging.publisher.AbstractMessagePublisher;
import com.inmobi.messaging.publisher.MessagePublisherFactory;

/**
 * Publishes integer sequence upto <code>maxSeq</code> on the <code>topic</code>,
 * each integer as message. 
 *
 * Prints out the publisher statistics at the end of the publishing.
 */
public class SeqGeneratorClient {

  public static void main(String[] args) throws Exception {
    if (args.length != 2) {
      System.err.println("Usage: SeqGeneratorClient <topic> <maxSeq>");
      return;
    }
    AbstractMessagePublisher publisher = 
        (AbstractMessagePublisher) MessagePublisherFactory.create();
    String topic = args[0];
    long maxSeq = Integer.parseInt(args[1]);
    publishMessages(publisher, topic, maxSeq);
    publisher.close();
    long invocation = publisher.getStats(topic).getInvocationCount();
    System.out.println("Total invocations: " + invocation);
    System.out.println("Total success: " +
        publisher.getStats(topic).getSuccessCount());
    System.out.println("Total unhandledExceptions: " +
        publisher.getStats(topic).getUnhandledExceptionCount());
  }

  static void publishMessages(AbstractMessagePublisher publisher, String topic,
      long maxSeq) throws InterruptedException {
    for (long seq = 1; seq <= maxSeq; seq++) {
      Message msg = new Message(ByteBuffer.wrap(Long.toString(seq).getBytes()));
      publisher.publish(topic, msg);
      Thread.sleep(1);
    }    
  }
}

@@ -874,16 +874,54 @@
 axSeq);%0A
+    waitToComplete(publisher, topic);%0A
     publ
@@ -1617,11 +1617,301 @@
     %0A  %7D
+%0A%0A  static void waitToComplete(AbstractMessagePublisher publisher,%0A      String topic) throws InterruptedException %7B%0A    while (publisher.getStats(topic).getInFlight() %3E 0) %7B%0A      System.out.println(%22Inflight: %22+ publisher.getStats(topic).getInFlight());%0A      Thread.sleep(100);%0A    %7D%0A  %7D
 %0A%7D%0A

Modify seqgen to wait for all messages to be sent before closing