*******************************************************************************/
package org.onap.dmaap.mr.client.impl;
+import java.io.File;
+import java.io.FileOutputStream;
import java.io.IOException;
import java.util.LinkedList;
import java.util.Properties;
import junit.framework.TestCase;
import org.junit.Test;
-
+import org.onap.dmaap.mr.client.MRClientFactory;
import org.onap.dmaap.mr.client.impl.MRConstants;
import org.onap.dmaap.mr.client.impl.MRConsumerImpl;
+import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants;
-public class MRConsumerImplTest extends TestCase
-{
+public class MRConsumerImplTest extends TestCase {
@Test
- public void testNullFilter () throws IOException
- {
- final LinkedList<String> hosts = new LinkedList<String> ();
- hosts.add ( "localhost:8080" );
- final MRConsumerImpl c = new MRConsumerImpl ( hosts, "topic", "cg", "cid", -1, -1, null, null, null );
- final String url = c.createUrlPath (MRConstants.makeConsumerUrl ( "localhost:8080", "topic", "cg", "cid","http" ), -1, -1 );
- assertEquals ("http://localhost:8080/events/" + "topic/cg/cid", url );
+ public void testNullFilter() throws IOException {
+ final LinkedList<String> hosts = new LinkedList<String>();
+ hosts.add("localhost:8080");
+ final MRConsumerImpl c = new MRConsumerImpl(hosts, "topic", "cg", "cid", -1, -1, null, null, null);
+ final String url = c.createUrlPath(MRConstants.makeConsumerUrl("localhost:8080", "topic", "cg", "cid", "http"),
+ -1, -1);
+ assertEquals("http://localhost:8080/events/" + "topic/cg/cid", url);
}
@Test
- public void testFilterWithNoTimeoutOrLimit () throws IOException
- {
- final LinkedList<String> hosts = new LinkedList<String> ();
- hosts.add ( "localhost:8080" );
- final MRConsumerImpl c = new MRConsumerImpl ( hosts, "topic", "cg", "cid", -1, -1, "filter", null, null );
- final String url = c.createUrlPath ( MRConstants.makeConsumerUrl ( "localhost:8080", "topic", "cg", "cid" ,"http"),-1, -1 );
- assertEquals ("http://localhost:8080/events/" + "topic/cg/cid?filter=filter", url );
+ public void testFilterWithNoTimeoutOrLimit() throws IOException {
+ final LinkedList<String> hosts = new LinkedList<String>();
+ hosts.add("localhost:8080");
+ final MRConsumerImpl c = new MRConsumerImpl(hosts, "topic", "cg", "cid", -1, -1, "filter", null, null);
+ final String url = c.createUrlPath(MRConstants.makeConsumerUrl("localhost:8080", "topic", "cg", "cid", "http"),
+ -1, -1);
+ assertEquals("http://localhost:8080/events/" + "topic/cg/cid?filter=filter", url);
}
@Test
- public void testTimeoutNoLimitNoFilter () throws IOException
- {
- final LinkedList<String> hosts = new LinkedList<String> ();
- hosts.add ( "localhost:8080" );
- final MRConsumerImpl c = new MRConsumerImpl ( hosts, "topic", "cg", "cid", 30000, -1, null, null, null );
- final String url = c.createUrlPath (MRConstants.makeConsumerUrl ( "localhost:8080", "topic", "cg", "cid","http" ), 30000, -1 );
- assertEquals ( "http://localhost:8080/events/" + "topic/cg/cid?timeout=30000", url );
+ public void testTimeoutNoLimitNoFilter() throws IOException {
+ final LinkedList<String> hosts = new LinkedList<String>();
+ hosts.add("localhost:8080");
+ final MRConsumerImpl c = new MRConsumerImpl(hosts, "topic", "cg", "cid", 30000, -1, null, null, null);
+ final String url = c.createUrlPath(MRConstants.makeConsumerUrl("localhost:8080", "topic", "cg", "cid", "http"),
+ 30000, -1);
+ assertEquals("http://localhost:8080/events/" + "topic/cg/cid?timeout=30000", url);
}
@Test
- public void testNoTimeoutWithLimitNoFilter () throws IOException
- {
- final LinkedList<String> hosts = new LinkedList<String> ();
- hosts.add ( "localhost:8080" );
- final MRConsumerImpl c = new MRConsumerImpl ( hosts, "topic", "cg", "cid", -1, 100, null, null, null );
- final String url = c.createUrlPath (MRConstants.makeConsumerUrl ( "localhost:8080", "topic", "cg", "cid","http" ), -1, 100 );
- assertEquals ( "http://localhost:8080/events/" + "topic/cg/cid?limit=100", url );
+ public void testNoTimeoutWithLimitNoFilter() throws IOException {
+ final LinkedList<String> hosts = new LinkedList<String>();
+ hosts.add("localhost:8080");
+ final MRConsumerImpl c = new MRConsumerImpl(hosts, "topic", "cg", "cid", -1, 100, null, null, null);
+ final String url = c.createUrlPath(MRConstants.makeConsumerUrl("localhost:8080", "topic", "cg", "cid", "http"),
+ -1, 100);
+ assertEquals("http://localhost:8080/events/" + "topic/cg/cid?limit=100", url);
}
@Test
- public void testWithTimeoutWithLimitWithFilter () throws IOException
- {
- final LinkedList<String> hosts = new LinkedList<String> ();
- hosts.add ( "localhost:8080" );
- final MRConsumerImpl c = new MRConsumerImpl ( hosts, "topic", "cg", "cid", 1000, 400, "f", null, null );
- final String url = c.createUrlPath (MRConstants.makeConsumerUrl ( "localhost:8080", "topic", "cg", "cid" ,"http"), 1000, 400 );
- assertEquals ("http://localhost:8080/events/" + "topic/cg/cid?timeout=1000&limit=400&filter=f", url );
+ public void testWithTimeoutWithLimitWithFilter() throws IOException {
+ final LinkedList<String> hosts = new LinkedList<String>();
+ hosts.add("localhost:8080");
+ final MRConsumerImpl c = new MRConsumerImpl(hosts, "topic", "cg", "cid", 1000, 400, "f", null, null);
+ final String url = c.createUrlPath(MRConstants.makeConsumerUrl("localhost:8080", "topic", "cg", "cid", "http"),
+ 1000, 400);
+ assertEquals("http://localhost:8080/events/" + "topic/cg/cid?timeout=1000&limit=400&filter=f", url);
}
@Test
- public void testFilterEncoding () throws IOException
- {
- final LinkedList<String> hosts = new LinkedList<String> ();
- hosts.add ( "localhost:8080" );
- final MRConsumerImpl c = new MRConsumerImpl ( hosts, "topic", "cg", "cid", -1, -1, "{ \"foo\"=\"bar\"bar\" }", null, null );
- final String url = c.createUrlPath (MRConstants.makeConsumerUrl ( "localhost:8080", "topic", "cg", "cid","http" ), -1, -1 );
- assertEquals ( "http://localhost:8080/events/" + "topic/cg/cid?filter=%7B+%22foo%22%3D%22bar%22bar%22+%7D", url );
+ public void testFilterEncoding() throws IOException {
+ final LinkedList<String> hosts = new LinkedList<String>();
+ hosts.add("localhost:8080");
+ final MRConsumerImpl c = new MRConsumerImpl(hosts, "topic", "cg", "cid", -1, -1, "{ \"foo\"=\"bar\"bar\" }",
+ null, null);
+ final String url = c.createUrlPath(MRConstants.makeConsumerUrl("localhost:8080", "topic", "cg", "cid", "http"),
+ -1, -1);
+ assertEquals("http://localhost:8080/events/" + "topic/cg/cid?filter=%7B+%22foo%22%3D%22bar%22bar%22+%7D", url);
}
-
+
@Test
- public void testFetchWithReturnConsumerResponse () throws IOException
- {
- final LinkedList<String> hosts = new LinkedList<String> ();
- hosts.add ( "localhost:8080" );
+ public void testFetchWithReturnConsumerResponse() throws IOException {
+ final LinkedList<String> hosts = new LinkedList<String>();
+ hosts.add("localhost:8080");
Properties properties = new Properties();
- properties.load(MRSimplerBatchConsumerTest.class.getClassLoader().getResourceAsStream("dme2/consumer.properties"));
+ properties.load(
+ MRSimplerBatchConsumerTest.class.getClassLoader().getResourceAsStream("dme2/consumer.properties"));
+
+ final MRConsumerImpl c = new MRConsumerImpl(hosts, "topic", "cg", "cid", -1, -1, "{ \"foo\"=\"bar\"bar\" }",
+ null, null);
+ assertNotNull(c.fetchWithReturnConsumerResponse());
+ c.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
+ assertNotNull(c.fetchWithReturnConsumerResponse());
+ c.setProtocolFlag(ProtocolTypeConstants.HTTPNOAUTH.getValue());
+ assertNotNull(c.fetchWithReturnConsumerResponse());
+ c.setProtocolFlag(ProtocolTypeConstants.AUTH_KEY.getValue());
+ assertNotNull(c.fetchWithReturnConsumerResponse());
+ assertTrue(true);
+ }
+
+ @Test
+ public void testFetch() throws Exception {
+ final LinkedList<String> hosts = new LinkedList<String>();
+ hosts.add("localhost:8080");
- final MRConsumerImpl c = new MRConsumerImpl ( hosts, "topic", "cg", "cid", -1, -1, "{ \"foo\"=\"bar\"bar\" }", null, null );
- c.fetchWithReturnConsumerResponse();
- c.setProtocolFlag("HTTPAAF");
- c.fetchWithReturnConsumerResponse();
+
+ Properties properties = new Properties();
+ properties.load(
+ MRSimplerBatchPublisherTest.class.getClassLoader().getResourceAsStream("dme2/consumer.properties"));
+
+ String routeFilePath = "dme2/preferredRoute.txt";
+
+ File file = new File(MRSimplerBatchPublisherTest.class.getClassLoader().getResource(routeFilePath).getFile());
+ properties.put("routeFilePath",
+ MRSimplerBatchPublisherTest.class.getClassLoader().getResource(routeFilePath).getFile());
+
+ File outFile = new File(file.getParent() + "/consumer_tmp.properties");
+ properties.store(new FileOutputStream(outFile), "");
+
+ MRClientFactory.prop=properties;
+ final MRConsumerImpl c = new MRConsumerImpl(hosts, "topic", "cg", "cid", -1, -1, "{ \"foo\"=\"bar\"bar\" }",
+ null, null);
+ c.setProps(properties);
+ try {
+ c.fetch();
+ } catch (Exception e) {
+ assertTrue(true);
+ }
+ c.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
+ assertNotNull(c.fetchWithReturnConsumerResponse());
+ c.setProtocolFlag(ProtocolTypeConstants.HTTPNOAUTH.getValue());
+ assertNotNull(c.fetchWithReturnConsumerResponse());
+ c.setProtocolFlag(ProtocolTypeConstants.AUTH_KEY.getValue());
+ assertNotNull(c.fetchWithReturnConsumerResponse());
assertTrue(true);
}
}
package org.onap.dmaap.mr.client.impl;
+import static org.junit.Assert.assertEquals;
+
import java.io.File;
+import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.List;
import org.onap.dmaap.mr.client.MRClientFactory;
import org.onap.dmaap.mr.client.MRPublisher.message;
import org.onap.dmaap.mr.client.response.MRPublisherResponse;
+import org.onap.dmaap.mr.test.clients.ProtocolTypeConstants;
public class MRSimplerBatchPublisherTest {
-
+
File outFile;
- @Before
- public void setUp() throws Exception {
+
+ public void setUp(String contentType) throws Exception {
Properties properties = new Properties();
- properties.load(MRSimplerBatchPublisherTest.class.getClassLoader().getResourceAsStream("dme2/producer.properties"));
-
- String routeFilePath="dme2/preferredRoute.txt";
-
+ properties.load(
+ MRSimplerBatchPublisherTest.class.getClassLoader().getResourceAsStream("dme2/producer.properties"));
+
+ String routeFilePath = "dme2/preferredRoute.txt";
+
File file = new File(MRSimplerBatchPublisherTest.class.getClassLoader().getResource(routeFilePath).getFile());
- properties.put("DME2preferredRouterFilePath", MRSimplerBatchPublisherTest.class.getClassLoader().getResource(routeFilePath).getFile());
-
+ properties.put("DME2preferredRouterFilePath",
+ MRSimplerBatchPublisherTest.class.getClassLoader().getResource(routeFilePath).getFile());
+ if (contentType != null) {
+ properties.put("contenttype", contentType);
+ }
outFile = new File(file.getParent() + "/producer_tmp.properties");
properties.store(new FileOutputStream(outFile), "");
}
@Test
- public void testSend() throws IOException, InterruptedException {
-
- final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher)MRClientFactory.createBatchingPublisher(outFile.getPath());
-
- //publish some messages
- final JSONObject msg1 = new JSONObject ();
- pub.send ( "MyPartitionKey", msg1.toString () );
+ public void testSend() throws Exception {
+
+ setUp(null);
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath());
+
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
+
+ final List<message> stuck = pub.close(1, TimeUnit.SECONDS);
+ Assert.assertEquals(1, stuck.size());
- final List<message> stuck = pub.close ( 1, TimeUnit.SECONDS );
- if ( stuck.size () > 0 ) {
- System.out.println( stuck.size() + " messages unsent" );
- }
- else
- {
- System.out.println ( "Clean exit; all messages sent." );
- }
-
-
}
@Test
- public void testSendBatchWithResponse() throws IOException, InterruptedException {
-
- final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher)MRClientFactory.createBatchingPublisher(outFile.getPath(), true);
-
- //publish some messages
- final JSONObject msg1 = new JSONObject ();
- pub.send ( "MyPartitionKey", msg1.toString () );
+ public void testSendBatchWithResponse() throws Exception {
+
+ setUp(null);
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
MRPublisherResponse pubResponse = new MRPublisherResponse();
pub.setPubResponse(pubResponse);
-
+
MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
+
+ }
+
+ @Test
+ public void testSendBatchWithResponseConText() throws Exception {
+
+ setUp("text/plain");
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath());
+
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
+
+ final List<message> stuck = pub.close(1, TimeUnit.SECONDS);
+ Assert.assertEquals(1, stuck.size());
+
+ }
+
+ @Test
+ public void testSendBatchWithResponseContCambria() throws Exception {
+
+ setUp("application/cambria-zip");
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath());
+
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
+
+ final List<message> stuck = pub.close(1, TimeUnit.SECONDS);
+ Assert.assertEquals(1, stuck.size());
+
+ }
+
+ @Test
+ public void testSendBatchWithResponseProtKey() throws Exception {
+
+ setUp(null);
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath());
+ pub.setProtocolFlag(ProtocolTypeConstants.AUTH_KEY.getValue());
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
+
+ final List<message> stuck = pub.close(1, TimeUnit.SECONDS);
+ Assert.assertEquals(1, stuck.size());
+
+ }
+
+ @Test
+ public void testSendBatchWithResponseProtAaf() throws Exception {
+
+ setUp(null);
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath());
+ pub.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
+
+ final List<message> stuck = pub.close(1, TimeUnit.SECONDS);
+ Assert.assertEquals(1, stuck.size());
+
+ }
+
+ @Test
+ public void testSendBatchWithResponseProtNoAuth() throws Exception {
+
+ setUp(null);
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath());
+ pub.setProtocolFlag(ProtocolTypeConstants.HTTPNOAUTH.getValue());
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
+
+ final List<message> stuck = pub.close(1, TimeUnit.SECONDS);
+ Assert.assertEquals(1, stuck.size());
+
+ }
+
+ @Test
+ public void testSendBatchWithResponsecontypeText() throws Exception {
+
+ setUp("text/plain");
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", "payload");
+ MRPublisherResponse pubResponse = new MRPublisherResponse();
+ pub.setPubResponse(pubResponse);
+
+ MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
+ Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
+
+ }
+
+ @Test
+ public void testSendBatchWithResponsecontypeCambria() throws Exception {
+
+ setUp("application/cambria-zip");
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", "payload");
+ MRPublisherResponse pubResponse = new MRPublisherResponse();
+ pub.setPubResponse(pubResponse);
+
+ MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
+ Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
+
+ }
+
+ @Test
+ public void testSendBatchWithResponsePrAuthKey() throws Exception {
+
+ setUp(null);
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+ pub.setProtocolFlag(ProtocolTypeConstants.AUTH_KEY.getValue());
+
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
+ MRPublisherResponse pubResponse = new MRPublisherResponse();
+ pub.setPubResponse(pubResponse);
+
+ MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
+ Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
+
+ }
+
+ @Test
+ public void testSendBatchWithResponsePrAaf() throws Exception {
+
+ setUp(null);
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+ pub.setProtocolFlag(ProtocolTypeConstants.AAF_AUTH.getValue());
+
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
+ MRPublisherResponse pubResponse = new MRPublisherResponse();
+ pub.setPubResponse(pubResponse);
+
+ MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
+ Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
+
+ }
+
+ @Test
+ public void testSendBatchWithResponsePrNoauth() throws Exception {
+
+ setUp(null);
+
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+ pub.setProtocolFlag(ProtocolTypeConstants.HTTPNOAUTH.getValue());
+
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
+ MRPublisherResponse pubResponse = new MRPublisherResponse();
+ pub.setPubResponse(pubResponse);
+
+ MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
+ Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
+
+ }
+
+ @Test
+ public void createPublisherResponse() throws Exception{
+ setUp(null);
+ MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+
+ MRPublisherResponse response=pub.createMRPublisherResponse("{\"message\": \"published the message\", \"status\": \"200\"}", new MRPublisherResponse());
+ assertEquals("200", response.getResponseCode());
+
+ }
+
+ @Test
+ public void createPublisherResponseSucc() throws Exception{
+ setUp(null);
+ MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+
+ MRPublisherResponse response=pub.createMRPublisherResponse("{\"fakemessage\": \"published the message\", \"fakestatus\": \"200\"}", new MRPublisherResponse());
+ assertEquals("200", response.getResponseCode());
+
+ }
+
+ @Test
+ public void createPublisherResponseError() throws Exception{
+ setUp(null);
+ MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+
+ MRPublisherResponse response=pub.createMRPublisherResponse("", new MRPublisherResponse());
+ assertEquals("400", response.getResponseCode());
}
Code Review / dmaap / messagerouter / dmaapclient.git / commitdiff