X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Forg%2Fonap%2Fdmaap%2Fmr%2Ftest%2Fclients%2FConsolePublisher.java;fp=src%2Ftest%2Fjava%2Forg%2Fonap%2Fdmaap%2Fmr%2Ftest%2Fclients%2FConsolePublisher.java;h=503af01aedcb00c35f9b31a6862dc950beb732a7;hb=72a9ab9e886cdeabc4b43418a7054a5796a0ff55;hp=0000000000000000000000000000000000000000;hpb=78ebc9a64fac6231e3e594200b9335a4c6372ed1;p=dmaap%2Fmessagerouter%2Fdmaapclient.git

diff --git a/src/test/java/org/onap/dmaap/mr/test/clients/ConsolePublisher.java b/src/test/java/org/onap/dmaap/mr/test/clients/ConsolePublisher.java
new file mode 100644
index 0000000..503af01
--- /dev/null
+++ b/src/test/java/org/onap/dmaap/mr/test/clients/ConsolePublisher.java
@@ -0,0 +1,90 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  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.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *  ============LICENSE_END=========================================================
+ *
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *
+ *******************************************************************************/
+
+package org.onap.dmaap.mr.test.clients;
+
+import org.onap.dmaap.mr.client.MRBatchingPublisher;
+import org.onap.dmaap.mr.client.MRClientFactory;
+import org.onap.dmaap.mr.client.MRPublisher.Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A simple publisher that reads from std in, sending each line as a message.
+ *
+ * @author author
+ */
+public class ConsolePublisher {
+
+    private static final Logger logger = LoggerFactory.getLogger(ConsolePublisher.class);
+
+    private ConsolePublisher() {
+    }
+
+    public static void main(String[] args) throws IOException //throws IOException, InterruptedException
+    {
+        // read the hosts(s) from the command line
+        final String hosts = args.length > 0 ? args[0] : "mr1.onap.com,mr2.onap.com,mr3.onap.com";
+
+        // read the topic name from the command line
+        final String topic = args.length > 1 ? args[1] : "TEST-TOPIC";
+
+        // read the topic name from the command line
+        final String partition = args.length > 2 ? args[2] : UUID.randomUUID().toString();
+
+        // set up some batch limits and the compression flag
+        final int maxBatchSize = 100;
+        final long maxAgeMs = 250;
+        final boolean withGzip = false;
+
+        // create our publisher
+        final MRBatchingPublisher pub = MRClientFactory.createBatchingPublisher(hosts, topic, maxBatchSize, maxAgeMs, withGzip);
+
+        final BufferedReader cin = new BufferedReader(new InputStreamReader(System.in));
+        try {
+            String line = null;
+            while ((line = cin.readLine()) != null) {
+                pub.send(partition, line);
+            }
+        } finally {
+            List<Message> leftovers = null;
+            try {
+                leftovers = pub.close(10, TimeUnit.SECONDS);
+            } catch (InterruptedException e) {
+                logger.error("Send on close interrupted.");
+                Thread.currentThread().interrupt();
+            }
+            for (Message m : leftovers) {
+                logger.error("Unsent message: " + m.fMsg);
+            }
+        }
+    }
+}