@@ -0,0 +1,2466 @@
+package com.inmobi.messaging;%0A%0Aimport static org.testng.Assert.assertEquals;%0A%0Aimport java.util.concurrent.CountDownLatch;%0A%0Aimport org.testng.annotations.AfterTest;%0Aimport org.testng.annotations.BeforeTest;%0Aimport org.testng.annotations.Test;%0A%0Aimport random.pkg.NtMultiServer;%0A%0Aimport com.inmobi.messaging.netty.ScribeMessagePublisher;%0A%0Apublic class TestMultiplePublisherThreads %7B%0A  private NtMultiServer server;%0A  private ScribeMessagePublisher publisher; %0A  %0A  @BeforeTest%0A  public void setUp() throws Exception %7B%0A    server = TestServerStarter.getServer();%0A    server.start();%0A    publisher = TestServerStarter.createPublisher();%0A  %7D%0A%0A  @AfterTest%0A  public void tearDown() %7B%0A    server.stop();%0A    if (publisher != null)%0A      publisher.close();%0A  %7D%0A  %0A  @Test%0A  public void multipleThreadSinglePublisher() throws Exception%7B%0A    String topic = %22test%22;%0A    CountDownLatch startLatch = new CountDownLatch(10);%0A    PublishThread p1 = new PublishThread(startLatch, topic, publisher);%0A    PublishThread p2 = new PublishThread(startLatch, topic, publisher);%0A    PublishThread p3 = new PublishThread(startLatch, topic, publisher);%0A    PublishThread p4 = new PublishThread(startLatch, topic, publisher);%0A    p1.start();%0A    p2.start();%0A    p3.start();%0A    p4.start();%0A    Thread.sleep(3000);%0A    p1.join();%0A    p2.join();%0A    p3.join();%0A    p4.join();%0A    System.out.println(publisher.getStats(topic));%0A    assertEquals(publisher.getStats(topic).getSuccessCount(), 4);%0A    assertEquals(publisher.getStats(topic).getRetryCount(),0);%0A    assertEquals(publisher.getStats(topic).getReconnectionCount(),0);%0A    System.out.println(%22Test multipleThreadSinglePublisher is done%22);    %0A  %7D%0A  %0A  private void doTest(String topic, ScribeMessagePublisher publisher) %7B%0A    publisher.publish(topic, new Message(%22msg1%22.getBytes()));%0A    while (publisher.getStats(topic).getInFlight() != 0) %7B%0A      try %7B%0A        Thread.sleep(100);%0A      %7D catch (InterruptedException e) %7B%0A        e.printStackTrace();%0A      %7D%0A    %7D%0A  %7D%0A  %0A  class PublishThread extends Thread %7B%0A    %0A    private String topic;%0A    private ScribeMessagePublisher publisher;%0A    private final CountDownLatch startLatch;%0A      %0A    PublishThread(CountDownLatch startLatch, String topic, ScribeMessagePublisher publisher)%7B%0A      this.startLatch= startLatch;%0A      this.topic = topic;%0A      this.publisher = publisher;%0A    %7D%0A      %0A    public void run()%7B%0A        doTest(topic, publisher);%0A        publisher.close();%0A    %7D%0A  %7D%0A%7D%0A

test for Initializing topic in synchrozied block