* ================================================================================
* Copyright © 2017 AT&T Intellectual Property. All rights reserved.
* ================================================================================
+ * Modifications Copyright © 2021 Orange.
+ * ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* ============LICENSE_END=========================================================
*
* ECOMP is a trademark and service mark of AT&T Intellectual Property.
- *
+ *
*******************************************************************************/
package org.onap.dmaap.mr.client.impl;
-import static org.junit.Assert.assertEquals;
+import org.json.JSONObject;
+import org.junit.Assert;
+import org.junit.Test;
+import org.onap.dmaap.mr.client.MRClientFactory;
+import org.onap.dmaap.mr.client.MRPublisher.Message;
+import org.onap.dmaap.mr.client.ProtocolType;
+import org.onap.dmaap.mr.client.response.MRPublisherResponse;
import java.io.File;
-import java.io.FileNotFoundException;
import java.io.FileOutputStream;
-import java.io.IOException;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
-import org.json.JSONObject;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-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;
+import static org.junit.Assert.assertEquals;
public class MRSimplerBatchPublisherTest {
- File outFile;
+ File outFile;
+
+ 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";
+
+ File file = new File(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 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());
+
+ }
+
+ @Test
+ 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 {
- public void setUp(String contentType) throws Exception {
- Properties properties = new Properties();
- properties.load(
- MRSimplerBatchPublisherTest.class.getClassLoader().getResourceAsStream("dme2/producer.properties"));
+ setUp("application/cambria-zip");
- String routeFilePath = "dme2/preferredRoute.txt";
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath());
- File file = new File(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), "");
- }
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
- @Test
- public void testSend() throws Exception {
+ final List<Message> stuck = pub.close(1, TimeUnit.SECONDS);
+ Assert.assertEquals(1, stuck.size());
- setUp(null);
+ }
- final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
- .createBatchingPublisher(outFile.getPath());
+ @Test
+ public void testSendBatchWithResponseProtKey() throws Exception {
- // publish some messages
- final JSONObject msg1 = new JSONObject();
- pub.send("MyPartitionKey", msg1.toString());
+ setUp(null);
- final List<message> stuck = pub.close(1, TimeUnit.SECONDS);
- Assert.assertEquals(1, stuck.size());
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath());
+ pub.setProtocolFlag(ProtocolType.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 testSendBatchWithResponse() throws Exception {
+ }
- setUp(null);
+ @Test
+ public void testSendBatchWithResponseProtAaf() throws Exception {
- final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
- .createBatchingPublisher(outFile.getPath(), true);
+ setUp(null);
- // publish some messages
- final JSONObject msg1 = new JSONObject();
- pub.send("MyPartitionKey", msg1.toString());
- MRPublisherResponse pubResponse = new MRPublisherResponse();
- pub.setPubResponse(pubResponse);
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath());
+ pub.setProtocolFlag(ProtocolType.AAF_AUTH.getValue());
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
- MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
- Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
+ final List<Message> stuck = pub.close(1, TimeUnit.SECONDS);
+ Assert.assertEquals(1, stuck.size());
- }
+ }
- @Test
- public void testSendBatchWithResponseConText() throws Exception {
+ @Test
+ public void testSendBatchWithResponseProtNoAuth() throws Exception {
- setUp("text/plain");
+ setUp(null);
- final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
- .createBatchingPublisher(outFile.getPath());
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath());
+ pub.setProtocolFlag(ProtocolType.HTTPNOAUTH.getValue());
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
- // 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);
- Assert.assertEquals(1, stuck.size());
+ }
- }
+ @Test
+ public void testSendBatchWithResponsecontypeText() throws Exception {
- @Test
- public void testSendBatchWithResponseContCambria() throws Exception {
+ setUp("text/plain");
- setUp("application/cambria-zip");
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
- final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
- .createBatchingPublisher(outFile.getPath());
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", "payload");
+ MRPublisherResponse pubResponse = new MRPublisherResponse();
+ pub.setPubResponse(pubResponse);
- // publish some messages
- final JSONObject msg1 = new JSONObject();
- pub.send("MyPartitionKey", msg1.toString());
+ MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
+ Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
- final List<message> stuck = pub.close(1, TimeUnit.SECONDS);
- Assert.assertEquals(1, stuck.size());
+ }
- }
+ @Test
+ public void testSendBatchWithResponsecontypeCambria() throws Exception {
- @Test
- public void testSendBatchWithResponseProtKey() throws Exception {
+ setUp("application/cambria-zip");
- setUp(null);
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
- 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());
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", "payload");
+ MRPublisherResponse pubResponse = new MRPublisherResponse();
+ pub.setPubResponse(pubResponse);
- final List<message> stuck = pub.close(1, TimeUnit.SECONDS);
- Assert.assertEquals(1, stuck.size());
+ MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
+ Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
- }
+ }
- @Test
- public void testSendBatchWithResponseProtAaf() throws Exception {
+ @Test
+ public void testSendBatchWithResponsePrAuthKey() throws Exception {
- setUp(null);
+ 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 MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+ pub.setProtocolFlag(ProtocolType.AUTH_KEY.getValue());
- final List<message> stuck = pub.close(1, TimeUnit.SECONDS);
- Assert.assertEquals(1, stuck.size());
+ // 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 testSendBatchWithResponseProtNoAuth() throws Exception {
+ }
- setUp(null);
+ @Test
+ public void testSendBatchWithResponsePrAaf() throws Exception {
- 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());
+ setUp(null);
- final List<message> stuck = pub.close(1, TimeUnit.SECONDS);
- Assert.assertEquals(1, stuck.size());
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+ pub.setProtocolFlag(ProtocolType.AAF_AUTH.getValue());
- }
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
+ MRPublisherResponse pubResponse = new MRPublisherResponse();
+ pub.setPubResponse(pubResponse);
- @Test
- public void testSendBatchWithResponsecontypeText() throws Exception {
+ MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
+ Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
- setUp("text/plain");
+ }
- final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
- .createBatchingPublisher(outFile.getPath(), true);
+ @Test
+ public void testSendBatchWithResponsePrNoauth() throws Exception {
- // publish some messages
- final JSONObject msg1 = new JSONObject();
- pub.send("MyPartitionKey", "payload");
- MRPublisherResponse pubResponse = new MRPublisherResponse();
- pub.setPubResponse(pubResponse);
+ setUp(null);
- MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
- Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
+ final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
+ pub.setProtocolFlag(ProtocolType.HTTPNOAUTH.getValue());
- }
+ // publish some messages
+ final JSONObject msg1 = new JSONObject();
+ pub.send("MyPartitionKey", msg1.toString());
+ MRPublisherResponse pubResponse = new MRPublisherResponse();
+ pub.setPubResponse(pubResponse);
- @Test
- public void testSendBatchWithResponsecontypeCambria() throws Exception {
+ MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
+ Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
- setUp("application/cambria-zip");
+ }
- final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
- .createBatchingPublisher(outFile.getPath(), true);
+ @Test
+ public void createPublisherResponse() throws Exception {
+ setUp(null);
+ 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 response = pub.createMRPublisherResponse("{\"message\": \"published the message\", \"status\": \"200\"}", new MRPublisherResponse());
+ assertEquals("200", response.getResponseCode());
- MRPublisherResponse mrPublisherResponse = pub.sendBatchWithResponse();
- Assert.assertEquals(1, mrPublisherResponse.getPendingMsgs());
+ }
- }
+ @Test
+ public void createPublisherResponseSucc() throws Exception {
+ setUp(null);
+ MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
+ .createBatchingPublisher(outFile.getPath(), true);
- @Test
- public void testSendBatchWithResponsePrAuthKey() throws Exception {
+ MRPublisherResponse response = pub.createMRPublisherResponse("{\"fakemessage\": \"published the message\", \"fakestatus\": \"200\"}", new MRPublisherResponse());
+ assertEquals("200", response.getResponseCode());
- setUp(null);
+ }
- final MRSimplerBatchPublisher pub = (MRSimplerBatchPublisher) MRClientFactory
- .createBatchingPublisher(outFile.getPath(), true);
- pub.setProtocolFlag(ProtocolTypeConstants.AUTH_KEY.getValue());
+ @Test
+ public void createPublisherResponseError() throws Exception {
+ setUp(null);
+ 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 testSendBatchWithResponsePrAaf() throws Exception {
+ MRPublisherResponse response = pub.createMRPublisherResponse("", new MRPublisherResponse());
+ assertEquals("400", response.getResponseCode());
- 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 / blobdiff