aboutsummaryrefslogtreecommitdiffstats
path: root/src/main/java/com/att/nsa
diff options
context:
space:
mode:
authorVarun Gudisena <vg411h@att.com>2017-08-31 10:44:28 -0500
committerVarun Gudisena <vg411h@att.com>2017-08-31 10:44:41 -0500
commit7d45c179879363222fcf49b30f75837f66d7f423 (patch)
treec5a344247515c1d8b74a6cc74bcea63541e4b46f /src/main/java/com/att/nsa
parentcc9de9bc6803212f0233e0e1bf06aa63fe8b7a6a (diff)
Revert package name changes
Reverted package name changes to avoid any potential issues. Renamed maven group id only. Issue-id: DMAAP-74 Change-Id: I36c2aef063050c265640b79e6dc0e8ab7add8d22 Signed-off-by: Varun Gudisena <vg411h@att.com>
Diffstat (limited to 'src/main/java/com/att/nsa')
-rw-r--r--src/main/java/com/att/nsa/mr/client/HostSelector.java191
-rw-r--r--src/main/java/com/att/nsa/mr/client/MRBatchingPublisher.java55
-rw-r--r--src/main/java/com/att/nsa/mr/client/MRClient.java66
-rw-r--r--src/main/java/com/att/nsa/mr/client/MRClientBuilders.java383
-rw-r--r--src/main/java/com/att/nsa/mr/client/MRClientFactory.java558
-rw-r--r--src/main/java/com/att/nsa/mr/client/MRConsumer.java54
-rw-r--r--src/main/java/com/att/nsa/mr/client/MRIdentityManager.java100
-rw-r--r--src/main/java/com/att/nsa/mr/client/MRPublisher.java93
-rw-r--r--src/main/java/com/att/nsa/mr/client/MRTopicManager.java183
-rw-r--r--src/main/java/com/att/nsa/mr/client/impl/Clock.java63
-rw-r--r--src/main/java/com/att/nsa/mr/client/impl/MRBaseClient.java393
-rw-r--r--src/main/java/com/att/nsa/mr/client/impl/MRBatchPublisher.java485
-rw-r--r--src/main/java/com/att/nsa/mr/client/impl/MRClientVersionInfo.java54
-rw-r--r--src/main/java/com/att/nsa/mr/client/impl/MRConstants.java180
-rw-r--r--src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java709
-rw-r--r--src/main/java/com/att/nsa/mr/client/impl/MRFormat.java49
-rw-r--r--src/main/java/com/att/nsa/mr/client/impl/MRMetaClient.java260
-rw-r--r--src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java939
-rw-r--r--src/main/java/com/att/nsa/mr/client/response/MRConsumerResponse.java60
-rw-r--r--src/main/java/com/att/nsa/mr/client/response/MRPublisherResponse.java66
-rw-r--r--src/main/java/com/att/nsa/mr/dme/client/DefaultLoggingFailoverFaultHandler.java50
-rw-r--r--src/main/java/com/att/nsa/mr/dme/client/HeaderReplyHandler.java63
-rw-r--r--src/main/java/com/att/nsa/mr/dme/client/PreferredRouteReplyHandler.java74
-rw-r--r--src/main/java/com/att/nsa/mr/dme/client/PreferredRouteRequestHandler.java54
-rw-r--r--src/main/java/com/att/nsa/mr/dme/client/SimpleExampleConsumer.java77
-rw-r--r--src/main/java/com/att/nsa/mr/dme/client/SimpleExamplePublisher.java135
-rw-r--r--src/main/java/com/att/nsa/mr/logging/MRAppender.java160
-rw-r--r--src/main/java/com/att/nsa/mr/test/clients/ConsolePublisher.java87
-rw-r--r--src/main/java/com/att/nsa/mr/test/clients/ProtocolTypeConstants.java46
-rw-r--r--src/main/java/com/att/nsa/mr/test/clients/SampleConsumer.java87
-rw-r--r--src/main/java/com/att/nsa/mr/test/clients/SamplePublisher.java86
-rw-r--r--src/main/java/com/att/nsa/mr/test/clients/SimpleExampleConsumer.java86
-rw-r--r--src/main/java/com/att/nsa/mr/test/clients/SimpleExampleConsumerWithReturnResponse.java90
-rw-r--r--src/main/java/com/att/nsa/mr/test/clients/SimpleExamplePublisher.java98
-rw-r--r--src/main/java/com/att/nsa/mr/test/clients/SimpleExamplePublisherWithResponse.java84
-rw-r--r--src/main/java/com/att/nsa/mr/test/support/MRBatchingPublisherMock.java184
-rw-r--r--src/main/java/com/att/nsa/mr/test/support/MRConsumerMock.java168
-rw-r--r--src/main/java/com/att/nsa/mr/tools/ApiKeyCommand.java134
-rw-r--r--src/main/java/com/att/nsa/mr/tools/AuthCommand.java69
-rw-r--r--src/main/java/com/att/nsa/mr/tools/ClusterCommand.java80
-rw-r--r--src/main/java/com/att/nsa/mr/tools/MRCommandContext.java100
-rw-r--r--src/main/java/com/att/nsa/mr/tools/MRTool.java49
-rw-r--r--src/main/java/com/att/nsa/mr/tools/MessageCommand.java128
-rw-r--r--src/main/java/com/att/nsa/mr/tools/TopicCommand.java209
-rw-r--r--src/main/java/com/att/nsa/mr/tools/TraceCommand.java118
45 files changed, 7457 insertions, 0 deletions
diff --git a/src/main/java/com/att/nsa/mr/client/HostSelector.java b/src/main/java/com/att/nsa/mr/client/HostSelector.java
new file mode 100644
index 0000000..de0e52c
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/HostSelector.java
@@ -0,0 +1,191 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.Random;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.Vector;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class HostSelector
+{
+ private final TreeSet<String> fBaseHosts;
+ private final DelayQueue<BlacklistEntry> fBlacklist;
+ private String fIdealHost;
+ private String fCurrentHost;
+ private static final Logger log = LoggerFactory.getLogger(HostSelector.class);
+
+ public HostSelector(String hostPart)
+ {
+ this(makeSet(hostPart), null);
+ }
+
+ public HostSelector(Collection<String> baseHosts)
+ {
+ this(baseHosts, null);
+ }
+
+ public HostSelector(Collection<String> baseHosts, String signature)
+ {
+ if (baseHosts.size() < 1)
+ {
+ throw new IllegalArgumentException("At least one host must be provided.");
+ }
+
+ this.fBaseHosts = new TreeSet(baseHosts);
+ this.fBlacklist = new DelayQueue();
+ this.fIdealHost = null;
+
+ if (signature == null) {
+ return;
+ }
+ int index = Math.abs(signature.hashCode()) % baseHosts.size();
+
+ Iterator it = this.fBaseHosts.iterator();
+ while (index-- > 0)
+ {
+ it.next();
+ }
+ this.fIdealHost = ((String)it.next());
+ }
+
+ public String selectBaseHost()
+ {
+ if (this.fCurrentHost == null)
+ {
+ makeSelection();
+ }
+ return this.fCurrentHost;
+ }
+
+ public void reportReachabilityProblem(long blacklistUnit, TimeUnit blacklistTimeUnit)
+ {
+ if (this.fCurrentHost == null)
+ {
+ log.warn("Reporting reachability problem, but no host is currently selected.");
+ }
+
+ if (blacklistUnit > 0L)
+ {
+ for (BlacklistEntry be : this.fBlacklist)
+ {
+ if (be.getHost().equals(this.fCurrentHost))
+ {
+ be.expireNow();
+ }
+ }
+
+ LinkedList devNull = new LinkedList();
+ this.fBlacklist.drainTo(devNull);
+
+ if (this.fCurrentHost != null)
+ {
+ this.fBlacklist.add(new BlacklistEntry(this.fCurrentHost, TimeUnit.MILLISECONDS.convert(blacklistUnit, blacklistTimeUnit)));
+ }
+ }
+ this.fCurrentHost = null;
+ }
+
+ private String makeSelection()
+ {
+ TreeSet workingSet = new TreeSet(this.fBaseHosts);
+
+ LinkedList devNull = new LinkedList();
+ this.fBlacklist.drainTo(devNull);
+ for (BlacklistEntry be : this.fBlacklist)
+ {
+ workingSet.remove(be.getHost());
+ }
+
+ if (workingSet.size() == 0)
+ {
+ log.warn("All hosts were blacklisted; reverting to full set of hosts.");
+ workingSet.addAll(this.fBaseHosts);
+ this.fCurrentHost = null;
+ }
+
+ String selection = null;
+ if ((this.fCurrentHost != null) && (workingSet.contains(this.fCurrentHost)))
+ {
+ selection = this.fCurrentHost;
+ }
+ else if ((this.fIdealHost != null) && (workingSet.contains(this.fIdealHost)))
+ {
+ selection = this.fIdealHost;
+ }
+ else
+ {
+ Vector v = new Vector(workingSet);
+ int index = Math.abs(new Random().nextInt()) % workingSet.size();
+ selection = (String)v.elementAt(index);
+ }
+
+ this.fCurrentHost = selection;
+ return this.fCurrentHost;
+ }
+
+ private static Set<String> makeSet(String s)
+ {
+ TreeSet set = new TreeSet();
+ set.add(s);
+ return set; }
+
+ private static class BlacklistEntry implements Delayed {
+ private final String fHost;
+ private long fExpireAtMs;
+
+ public BlacklistEntry(String host, long delayMs) {
+ this.fHost = host;
+ this.fExpireAtMs = (System.currentTimeMillis() + delayMs);
+ }
+
+ public void expireNow()
+ {
+ this.fExpireAtMs = 0L;
+ }
+
+ public String getHost()
+ {
+ return this.fHost;
+ }
+
+ public int compareTo(Delayed o)
+ {
+ Long thisDelay = Long.valueOf(getDelay(TimeUnit.MILLISECONDS));
+ return thisDelay.compareTo(Long.valueOf(o.getDelay(TimeUnit.MILLISECONDS)));
+ }
+
+ public long getDelay(TimeUnit unit)
+ {
+ long remainingMs = this.fExpireAtMs - System.currentTimeMillis();
+ return unit.convert(remainingMs, TimeUnit.MILLISECONDS);
+ }
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/client/MRBatchingPublisher.java b/src/main/java/com/att/nsa/mr/client/MRBatchingPublisher.java
new file mode 100644
index 0000000..875b5a3
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/MRBatchingPublisher.java
@@ -0,0 +1,55 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import com.att.nsa.mr.client.response.MRPublisherResponse;
+
+/**
+ * A MR batching publisher is a publisher with additional functionality
+ * for managing delayed sends.
+ *
+ * @author author
+ *
+ */
+public interface MRBatchingPublisher extends MRPublisher
+{
+ /**
+ * Get the number of messages that have not yet been sent.
+ * @return the number of pending messages
+ */
+ int getPendingMessageCount ();
+
+ /**
+ * Close this publisher, sending any remaining messages.
+ * @param timeout an amount of time to wait for unsent messages to be sent
+ * @param timeoutUnits the time unit for the timeout arg
+ * @return a list of any unsent messages after the timeout
+ * @throws IOException exception
+ * @throws InterruptedException exception
+ */
+ List<message> close ( long timeout, TimeUnit timeoutUnits ) throws IOException, InterruptedException;
+
+ MRPublisherResponse sendBatchWithResponse ();
+}
diff --git a/src/main/java/com/att/nsa/mr/client/MRClient.java b/src/main/java/com/att/nsa/mr/client/MRClient.java
new file mode 100644
index 0000000..f3d5c88
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/MRClient.java
@@ -0,0 +1,66 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client;
+
+import org.slf4j.Logger;
+
+
+public interface MRClient
+{
+ /**
+ * An exception at the MR layer. This is used when the HTTP transport
+ * layer returns a success code but the transaction is not completed as expected.
+ */
+ public class MRApiException extends Exception
+ {
+ public MRApiException ( String msg ) { super ( msg ); }
+ public MRApiException ( String msg, Throwable t ) { super ( msg, t ); }
+ private static final long serialVersionUID = 1L;
+ }
+
+ /**
+ * Optionally set the Logger to use
+ * @param log log
+ */
+ void logTo ( Logger log );
+
+ /**
+ * Set the API credentials for this client connection. Subsequent calls will
+ * include authentication headers.who i
+ */
+ /**
+ * @param apiKey apikey
+ * @param apiSecret apisec
+ */
+ void setApiCredentials ( String apiKey, String apiSecret );
+
+ /**
+ * Remove API credentials, if any, on this connection. Subsequent calls will not include
+ * authentication headers.
+ */
+ void clearApiCredentials ();
+
+ /**
+ * Close this connection. Some client interfaces have additional close capability.
+ */
+ void close ();
+}
diff --git a/src/main/java/com/att/nsa/mr/client/MRClientBuilders.java b/src/main/java/com/att/nsa/mr/client/MRClientBuilders.java
new file mode 100644
index 0000000..00c8915
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/MRClientBuilders.java
@@ -0,0 +1,383 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client;
+
+import java.net.MalformedURLException;
+import java.util.Collection;
+import java.util.TreeSet;
+import java.util.UUID;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.nsa.mr.client.impl.MRConsumerImpl;
+import com.att.nsa.mr.client.impl.MRMetaClient;
+import com.att.nsa.mr.client.impl.MRSimplerBatchPublisher;
+
+/**
+ * A collection of builders for various types of MR API clients
+ *
+ * @author author
+ */
+public class MRClientBuilders
+{
+ /**
+ * A builder for a topic Consumer
+ * @author author
+ */
+ public static class ConsumerBuilder
+ {
+ /**
+ * Construct a consumer builder.
+ */
+ public ConsumerBuilder () {}
+
+ /**
+ * Set the host list
+ * @param hostList a comma-separated list of hosts to use to connect to MR
+ * @return this builder
+ */
+ public ConsumerBuilder usingHosts ( String hostList ) { return usingHosts ( MRConsumerImpl.stringToList(hostList) ); }
+
+ /**
+ * Set the host list
+ * @param hostSet a set of hosts to use to connect to MR
+ * @return this builder
+ */
+ public ConsumerBuilder usingHosts ( Collection<String> hostSet ) { fHosts = hostSet; return this; }
+
+ /**
+ * Set the topic
+ * @param topic the name of the topic to consume
+ * @return this builder
+ */
+ public ConsumerBuilder onTopic ( String topic ) { fTopic=topic; return this; }
+
+ /**
+ * Set the consumer's group and ID
+ * @param consumerGroup The name of the consumer group this consumer is part of
+ * @param consumerId The unique id of this consumer in its group
+ * @return this builder
+ */
+ public ConsumerBuilder knownAs ( String consumerGroup, String consumerId ) { fGroup = consumerGroup; fId = consumerId; return this; }
+
+ /**
+ * Set the API key and secret for this client.
+ * @param apiKey
+ * @param apiSecret
+ * @return this builder
+ */
+ public ConsumerBuilder authenticatedBy ( String apiKey, String apiSecret ) { fApiKey = apiKey; fApiSecret = apiSecret; return this; }
+
+ /**
+ * Set the server side timeout
+ * @param timeoutMs The amount of time in milliseconds that the server should keep the connection open while waiting for message traffic.
+ * @return this builder
+ */
+ public ConsumerBuilder waitAtServer ( int timeoutMs ) { fTimeoutMs = timeoutMs; return this; };
+
+ /**
+ * Set the maximum number of messages to receive per transaction
+ * @param limit The maximum number of messages to receive from the server in one transaction.
+ * @return this builder
+ */
+ public ConsumerBuilder receivingAtMost ( int limit ) { fLimit = limit; return this; };
+
+ /**
+ * Set a filter to use on the server
+ * @param filter a Highland Park standard library filter encoded in JSON
+ * @return this builder
+ */
+ public ConsumerBuilder withServerSideFilter ( String filter ) { fFilter = filter; return this; }
+
+ /**
+ * Build the consumer
+ * @return a consumer
+ */
+ public MRConsumer build ()
+ {
+ if ( fHosts == null || fHosts.size() == 0 || fTopic == null )
+ {
+ throw new IllegalArgumentException ( "You must provide at least one host and a topic name." );
+ }
+
+ if ( fGroup == null )
+ {
+ fGroup = UUID.randomUUID ().toString ();
+ fId = "0";
+ log.info ( "Creating non-restartable client with group " + fGroup + " and ID " + fId + "." );
+ }
+
+ if ( sfConsumerMock != null ) return sfConsumerMock;
+ try {
+ return new MRConsumerImpl ( fHosts, fTopic, fGroup, fId, fTimeoutMs, fLimit, fFilter, fApiKey, fApiSecret );
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Collection<String> fHosts = null;
+ private String fTopic = null;
+ private String fGroup = null;
+ private String fId = null;
+ private String fApiKey = null;
+ private String fApiSecret = null;
+ private int fTimeoutMs = -1;
+ private int fLimit = -1;
+ private String fFilter = null;
+ }
+
+ /*************************************************************************/
+ /*************************************************************************/
+ /*************************************************************************/
+
+ /**
+ * A publisher builder
+ * @author author
+ */
+ public static class PublisherBuilder
+ {
+ public PublisherBuilder () {}
+
+ /**
+ * Set the MR/UEB host(s) to use
+ * @param hostlist The host(s) used in the URL to MR. Can be "host:port", can be multiple comma-separated entries.
+ * @return this builder
+ */
+ public PublisherBuilder usingHosts ( String hostlist ) { return usingHosts ( MRConsumerImpl.stringToList(hostlist) ); }
+
+ /**
+ * Set the MR/UEB host(s) to use
+ * @param hostSet The host(s) used in the URL to MR. Can be "host:port"
+ * @return this builder
+ */
+ public PublisherBuilder usingHosts ( String[] hostSet )
+ {
+ final TreeSet<String> hosts = new TreeSet<String> ();
+ for ( String hp : hostSet )
+ {
+ hosts.add ( hp );
+ }
+ return usingHosts ( hosts );
+ }
+
+ /**
+ * Set the MR/UEB host(s) to use
+ * @param hostlist The host(s) used in the URL to MR. Can be "host:port".
+ * @return this builder
+ */
+ public PublisherBuilder usingHosts ( Collection<String> hostlist ) { fHosts=hostlist; return this; }
+
+ /**
+ * Set the topic to publish on
+ * @param topic The topic on which to publish messages.
+ * @return this builder
+ */
+ public PublisherBuilder onTopic ( String topic ) { fTopic = topic; return this; }
+
+ /**
+ * Batch message sends with the given limits.
+ * @param messageCount The largest set of messages to batch.
+ * @param ageInMs The maximum age of a message waiting in a batch.
+ * @return this builder
+ */
+ public PublisherBuilder limitBatch ( int messageCount, int ageInMs ) { fMaxBatchSize = messageCount; fMaxBatchAgeMs = ageInMs; return this; }
+
+ /**
+ * Compress transactions
+ * @return this builder
+ */
+ public PublisherBuilder withCompresion () { return enableCompresion(true); }
+
+ /**
+ * Do not compress transactions
+ * @return this builder
+ */
+ public PublisherBuilder withoutCompresion () { return enableCompresion(false); }
+
+ /**
+ * Set the compression option
+ * @param compress true to gzip compress transactions
+ * @return this builder
+ */
+ public PublisherBuilder enableCompresion ( boolean compress ) { fCompress = compress; return this; }
+
+ /**
+ * Set the API key and secret for this client.
+ * @param apiKey
+ * @param apiSecret
+ * @return this builder
+ */
+ public PublisherBuilder authenticatedBy ( String apiKey, String apiSecret ) { fApiKey = apiKey; fApiSecret = apiSecret; return this; }
+
+ /**
+ * Build the publisher
+ * @return a batching publisher
+ */
+ public MRBatchingPublisher build ()
+ {
+ if ( fHosts == null || fHosts.size() == 0 || fTopic == null )
+ {
+ throw new IllegalArgumentException ( "You must provide at least one host and a topic name." );
+ }
+
+ if ( sfPublisherMock != null ) return sfPublisherMock;
+
+ final MRSimplerBatchPublisher pub = new MRSimplerBatchPublisher.Builder ().
+ againstUrls ( fHosts ).
+ onTopic ( fTopic ).
+ batchTo ( fMaxBatchSize, fMaxBatchAgeMs ).
+ compress ( fCompress ).
+ build ();
+ if ( fApiKey != null )
+ {
+ pub.setApiCredentials ( fApiKey, fApiSecret );
+ }
+ return pub;
+ }
+
+ private Collection<String> fHosts = null;
+ private String fTopic = null;
+ private int fMaxBatchSize = 1;
+ private int fMaxBatchAgeMs = 1;
+ private boolean fCompress = false;
+ private String fApiKey = null;
+ private String fApiSecret = null;
+ }
+
+ /**
+ * A builder for an identity manager
+ * @author author
+ */
+ public static class IdentityManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRIdentityManager>
+ {
+ /**
+ * Construct an identity manager builder.
+ */
+ public IdentityManagerBuilder () {}
+
+ @Override
+ protected MRIdentityManager constructClient ( Collection<String> hosts ) { try {
+ return new MRMetaClient ( hosts );
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ } }
+ }
+
+ /**
+ * A builder for a topic manager
+ * @author author
+ */
+ public static class TopicManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRTopicManager>
+ {
+ /**
+ * Construct an topic manager builder.
+ */
+ public TopicManagerBuilder () {}
+
+ @Override
+ protected MRTopicManager constructClient ( Collection<String> hosts ) { try {
+ return new MRMetaClient ( hosts );
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ } }
+ }
+
+ /**
+ * Inject a consumer. Used to support unit tests.
+ * @param cc
+ */
+ public static void $testInject ( MRConsumer cc )
+ {
+ sfConsumerMock = cc;
+ }
+
+ /**
+ * Inject a publisher. Used to support unit tests.
+ * @param pub
+ */
+ public static void $testInject ( MRBatchingPublisher pub )
+ {
+ sfPublisherMock = pub;
+ }
+
+ static MRConsumer sfConsumerMock = null;
+ static MRBatchingPublisher sfPublisherMock = null;
+
+ /**
+ * A builder for an identity manager
+ * @author author
+ */
+ public static abstract class AbstractAuthenticatedManagerBuilder<T extends MRClient>
+ {
+ /**
+ * Construct an identity manager builder.
+ */
+ public AbstractAuthenticatedManagerBuilder () {}
+
+ /**
+ * Set the host list
+ * @param hostList a comma-separated list of hosts to use to connect to MR
+ * @return this builder
+ */
+ public AbstractAuthenticatedManagerBuilder<T> usingHosts ( String hostList ) { return usingHosts ( MRConsumerImpl.stringToList(hostList) ); }
+
+ /**
+ * Set the host list
+ * @param hostSet a set of hosts to use to connect to MR
+ * @return this builder
+ */
+ public AbstractAuthenticatedManagerBuilder<T> usingHosts ( Collection<String> hostSet ) { fHosts = hostSet; return this; }
+
+ /**
+ * Set the API key and secret for this client.
+ * @param apiKey
+ * @param apiSecret
+ * @return this builder
+ */
+ public AbstractAuthenticatedManagerBuilder<T> authenticatedBy ( String apiKey, String apiSecret ) { fApiKey = apiKey; fApiSecret = apiSecret; return this; }
+
+ /**
+ * Build the consumer
+ * @return a consumer
+ */
+ public T build ()
+ {
+ if ( fHosts == null || fHosts.size() == 0 )
+ {
+ throw new IllegalArgumentException ( "You must provide at least one host and a topic name." );
+ }
+
+ final T mgr = constructClient ( fHosts );
+ mgr.setApiCredentials ( fApiKey, fApiSecret );
+ return mgr;
+ }
+
+ protected abstract T constructClient ( Collection<String> hosts );
+
+ private Collection<String> fHosts = null;
+ private String fApiKey = null;
+ private String fApiSecret = null;
+ }
+
+ private static final Logger log = LoggerFactory.getLogger ( MRClientBuilders.class );
+}
diff --git a/src/main/java/com/att/nsa/mr/client/MRClientFactory.java b/src/main/java/com/att/nsa/mr/client/MRClientFactory.java
new file mode 100644
index 0000000..59e472c
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/MRClientFactory.java
@@ -0,0 +1,558 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+import java.util.TreeSet;
+import java.util.UUID;
+
+import javax.ws.rs.core.MultivaluedMap;
+
+import com.att.nsa.mr.client.impl.MRConsumerImpl;
+import com.att.nsa.mr.client.impl.MRMetaClient;
+import com.att.nsa.mr.client.impl.MRSimplerBatchPublisher;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+/**
+ * A factory for MR clients.<br/>
+ * <br/>
+ * Use caution selecting a consumer creator factory. If the call doesn't accept a consumer group name, then it creates
+ * a consumer that is not restartable. That is, if you stop your process and start it again, your client will NOT receive
+ * any missed messages on the topic. If you need to ensure receipt of missed messages, then you must use a consumer that's
+ * created with a group name and ID. (If you create multiple consumer processes using the same group, load is split across
+ * them. Be sure to use a different ID for each instance.)<br/>
+ * <br/>
+ * Publishers
+ *
+ * @author author
+ */
+public class MRClientFactory
+{
+ public static MultivaluedMap<String, Object> HTTPHeadersMap;
+ public static Map<String, String> DME2HeadersMap;
+ public static String routeFilePath;
+
+ public static FileReader routeReader;
+
+ public static FileWriter routeWriter= null;
+ public static Properties prop=null;
+ //routeReader= new FileReader(new File (routeFilePath));
+ //props= new Properties();
+ /**
+ * Create a consumer instance with the default timeout and no limit
+ * on messages returned. This consumer operates as an independent consumer (i.e., not in a group) and is NOT re-startable
+ * across sessions.
+ *
+ * @param hostList A comma separated list of hosts to use to connect to MR.
+ * You can include port numbers (3904 is the default). For example, "hostname:8080,"
+ *
+ * @param topic The topic to consume
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer ( String hostList, String topic )
+ {
+ return createConsumer ( MRConsumerImpl.stringToList(hostList), topic );
+ }
+
+ /**
+ * Create a consumer instance with the default timeout and no limit
+ * on messages returned. This consumer operates as an independent consumer (i.e., not in a group) and is NOT re-startable
+ * across sessions.
+ *
+ * @param hostSet The host used in the URL to MR. Entries can be "host:port".
+ * @param topic The topic to consume
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer ( Collection<String> hostSet, String topic )
+ {
+ return createConsumer ( hostSet, topic, null );
+ }
+
+ /**
+ * Create a consumer instance with server-side filtering, the default timeout, and no limit
+ * on messages returned. This consumer operates as an independent consumer (i.e., not in a group) and is NOT re-startable
+ * across sessions.
+ *
+ * @param hostSet The host used in the URL to MR. Entries can be "host:port".
+ * @param topic The topic to consume
+ * @param filter a filter to use on the server side
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer ( Collection<String> hostSet, String topic, String filter )
+ {
+ return createConsumer ( hostSet, topic, UUID.randomUUID ().toString (), "0", -1, -1, filter, null, null );
+ }
+
+ /**
+ * Create a consumer instance with the default timeout, and no limit
+ * on messages returned. This consumer can operate in a logical group and is re-startable
+ * across sessions when you use the same group and ID on restart.
+ *
+ * @param hostSet The host used in the URL to MR. Entries can be "host:port".
+ * @param topic The topic to consume
+ * @param consumerGroup The name of the consumer group this consumer is part of
+ * @param consumerId The unique id of this consume in its group
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer ( Collection<String> hostSet, final String topic, final String consumerGroup, final String consumerId )
+ {
+ return createConsumer ( hostSet, topic, consumerGroup, consumerId, -1, -1 );
+ }
+
+ /**
+ * Create a consumer instance with the default timeout, and no limit
+ * on messages returned. This consumer can operate in a logical group and is re-startable
+ * across sessions when you use the same group and ID on restart.
+ *
+ * @param hostSet The host used in the URL to MR. Entries can be "host:port".
+ * @param topic The topic to consume
+ * @param consumerGroup The name of the consumer group this consumer is part of
+ * @param consumerId The unique id of this consume in its group
+ * @param timeoutMs The amount of time in milliseconds that the server should keep the connection open while waiting for message traffic. Use -1 for default timeout.
+ * @param limit A limit on the number of messages returned in a single call. Use -1 for no limit.
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer ( Collection<String> hostSet, final String topic, final String consumerGroup, final String consumerId, int timeoutMs, int limit)
+ {
+ return createConsumer ( hostSet, topic, consumerGroup, consumerId, timeoutMs, limit, null, null, null );
+ }
+
+ /**
+ * Create a consumer instance with the default timeout, and no limit
+ * on messages returned. This consumer can operate in a logical group and is re-startable
+ * across sessions when you use the same group and ID on restart. This consumer also uses
+ * server-side filtering.
+ *
+ * @param hostList A comma separated list of hosts to use to connect to MR.
+ * You can include port numbers (3904 is the default). For example, "ueb01hydc.it.att.com:8080,ueb02hydc.it.att.com"
+ * @param topic The topic to consume
+ * @param consumerGroup The name of the consumer group this consumer is part of
+ * @param consumerId The unique id of this consume in its group
+ * @param timeoutMs The amount of time in milliseconds that the server should keep the connection open while waiting for message traffic. Use -1 for default timeout.
+ * @param limit A limit on the number of messages returned in a single call. Use -1 for no limit.
+ * @param filter A Highland Park filter expression using only built-in filter components. Use null for "no filter".
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer ( String hostList, final String topic, final String consumerGroup,
+ final String consumerId, int timeoutMs, int limit, String filter, String apiKey, String apiSecret )
+ {
+ return createConsumer ( MRConsumerImpl.stringToList(hostList), topic, consumerGroup,
+ consumerId, timeoutMs, limit, filter, apiKey, apiSecret );
+ }
+
+ /**
+ * Create a consumer instance with the default timeout, and no limit
+ * on messages returned. This consumer can operate in a logical group and is re-startable
+ * across sessions when you use the same group and ID on restart. This consumer also uses
+ * server-side filtering.
+ *
+ * @param hostSet The host used in the URL to MR. Entries can be "host:port".
+ * @param topic The topic to consume
+ * @param consumerGroup The name of the consumer group this consumer is part of
+ * @param consumerId The unique id of this consume in its group
+ * @param timeoutMs The amount of time in milliseconds that the server should keep the connection open while waiting for message traffic. Use -1 for default timeout.
+ * @param limit A limit on the number of messages returned in a single call. Use -1 for no limit.
+ * @param filter A Highland Park filter expression using only built-in filter components. Use null for "no filter".
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer ( Collection<String> hostSet, final String topic, final String consumerGroup,
+ final String consumerId, int timeoutMs, int limit, String filter, String apiKey, String apiSecret )
+ {
+ if ( MRClientBuilders.sfConsumerMock != null ) return MRClientBuilders.sfConsumerMock;
+ try {
+ return new MRConsumerImpl ( hostSet, topic, consumerGroup, consumerId, timeoutMs, limit, filter, apiKey, apiSecret );
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /*************************************************************************/
+ /*************************************************************************/
+ /*************************************************************************/
+
+ /**
+ * Create a publisher that sends each message (or group of messages) immediately. Most
+ * applications should favor higher latency for much higher message throughput and the
+ * "simple publisher" is not a good choice.
+ *
+ * @param hostlist The host used in the URL to MR. Can be "host:port", can be multiple comma-separated entries.
+ * @param topic The topic on which to publish messages.
+ * @return a publisher
+ */
+ public static MRBatchingPublisher createSimplePublisher ( String hostlist, String topic )
+ {
+ return createBatchingPublisher ( hostlist, topic, 1, 1 );
+ }
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher to
+ * send the last batch and ensure a clean shutdown. Message payloads are not compressed.
+ *
+ * @param hostlist The host used in the URL to MR. Can be "host:port", can be multiple comma-separated entries.
+ * @param topic The topic on which to publish messages.
+ * @param maxBatchSize The largest set of messages to batch
+ * @param maxAgeMs The maximum age of a message waiting in a batch
+ *
+ * @return a publisher
+ */
+ public static MRBatchingPublisher createBatchingPublisher ( String hostlist, String topic, int maxBatchSize, long maxAgeMs )
+ {
+ return createBatchingPublisher ( hostlist, topic, maxBatchSize, maxAgeMs, false );
+ }
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher to
+ * send the last batch and ensure a clean shutdown.
+ *
+ * @param hostlist The host used in the URL to MR. Can be "host:port", can be multiple comma-separated entries.
+ * @param topic The topic on which to publish messages.
+ * @param maxBatchSize The largest set of messages to batch
+ * @param maxAgeMs The maximum age of a message waiting in a batch
+ * @param compress use gzip compression
+ *
+ * @return a publisher
+ */
+ public static MRBatchingPublisher createBatchingPublisher ( String hostlist, String topic, int maxBatchSize, long maxAgeMs, boolean compress )
+ {
+ return createBatchingPublisher ( MRConsumerImpl.stringToList(hostlist), topic, maxBatchSize, maxAgeMs, compress );
+ }
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher to
+ * send the last batch and ensure a clean shutdown.
+ *
+ * @param hostSet A set of hosts to be used in the URL to MR. Can be "host:port". Use multiple entries to enable failover.
+ * @param topic The topic on which to publish messages.
+ * @param maxBatchSize The largest set of messages to batch
+ * @param maxAgeMs The maximum age of a message waiting in a batch
+ * @param compress use gzip compression
+ *
+ * @return a publisher
+ */
+ public static MRBatchingPublisher createBatchingPublisher ( String[] hostSet, String topic, int maxBatchSize, long maxAgeMs, boolean compress )
+ {
+ final TreeSet<String> hosts = new TreeSet<String> ();
+ for ( String hp : hostSet )
+ {
+ hosts.add ( hp );
+ }
+ return createBatchingPublisher ( hosts, topic, maxBatchSize, maxAgeMs, compress );
+ }
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher to
+ * send the last batch and ensure a clean shutdown.
+ *
+ * @param hostSet A set of hosts to be used in the URL to MR. Can be "host:port". Use multiple entries to enable failover.
+ * @param topic The topic on which to publish messages.
+ * @param maxBatchSize The largest set of messages to batch
+ * @param maxAgeMs The maximum age of a message waiting in a batch
+ * @param compress use gzip compression
+ *
+ * @return a publisher
+ */
+ public static MRBatchingPublisher createBatchingPublisher ( Collection<String> hostSet, String topic, int maxBatchSize, long maxAgeMs, boolean compress )
+ {
+ return new MRSimplerBatchPublisher.Builder ().
+ againstUrls ( hostSet ).
+ onTopic ( topic ).
+ batchTo ( maxBatchSize, maxAgeMs ).
+ compress ( compress ).
+ build ();
+ }
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher to
+ * send the last batch and ensure a clean shutdown.
+ * @param host A host to be used in the URL to MR. Can be "host:port". Use multiple entries to enable failover.
+ * @param topic The topic on which to publish messages.
+ * @param username username
+ * @param password password
+ * @param maxBatchSize The largest set of messages to batch
+ * @param maxAgeMs The maximum age of a message waiting in a batch
+ * @param compress use gzip compression
+ * @param protocolFlag http auth or ueb auth or dme2 method
+ * @param producerFilePath all properties for publisher
+ * @return MRBatchingPublisher obj
+ */
+ public static MRBatchingPublisher createBatchingPublisher ( String host, String topic, final String username, final String password, int maxBatchSize, long maxAgeMs, boolean compress, String protocolFlag, String producerFilePath )
+ {
+ MRSimplerBatchPublisher pub = new MRSimplerBatchPublisher.Builder ().
+ againstUrls(MRConsumerImpl.stringToList(host)).
+ onTopic ( topic ).
+ batchTo ( maxBatchSize, maxAgeMs ).
+ compress ( compress ).
+ build ();
+
+ pub.setHost(host);
+ pub.setUsername(username);
+ pub.setPassword(password);
+ pub.setProtocolFlag(protocolFlag);
+ pub.setProducerFilePath(producerFilePath);
+ return pub;
+ }
+
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher to
+ * send the last batch and ensure a clean shutdown
+ * @param producerFilePath set all properties for publishing message
+ * @return MRBatchingPublisher obj
+ * @throws FileNotFoundException exc
+ * @throws IOException ioex
+ */
+ public static MRBatchingPublisher createBatchingPublisher ( final String producerFilePath ) throws FileNotFoundException,IOException {
+ FileReader reader = new FileReader(new File (producerFilePath));
+ Properties props = new Properties();
+ props.load(reader);
+ MRSimplerBatchPublisher pub = new MRSimplerBatchPublisher.Builder ().
+ againstUrls(MRConsumerImpl.stringToList(props.getProperty("host"))).
+ onTopic ( props.getProperty("topic") ).
+ batchTo (Integer.parseInt(props.getProperty("maxBatchSize")), Integer.parseInt(props.getProperty("maxAgeMs").toString())).
+ compress (Boolean.parseBoolean(props.getProperty("compress"))).
+ httpThreadTime(Integer.parseInt(props.getProperty("MessageSentThreadOccurance"))).
+ build ();
+ pub.setHost(props.getProperty("host"));
+ if(props.getProperty("TransportType").equalsIgnoreCase(ProtocolTypeConstants.AUTH_KEY.getValue())){
+
+ pub.setAuthKey(props.getProperty("authKey"));
+ pub.setAuthDate(props.getProperty("authDate"));
+ pub.setUsername(props.getProperty("username"));
+ pub.setPassword(props.getProperty("password"));
+ }else{
+ pub.setUsername(props.getProperty("username"));
+ pub.setPassword(props.getProperty("password"));
+ }
+ pub.setProducerFilePath(producerFilePath);
+ pub.setProtocolFlag(props.getProperty("TransportType"));
+ pub.setProps(props);
+ routeFilePath=props.getProperty("DME2preferredRouterFilePath");
+ routeReader= new FileReader(new File (routeFilePath));
+ prop= new Properties();
+ File fo= new File(routeFilePath);
+ if(!fo.exists()){
+ routeWriter=new FileWriter(new File(routeFilePath));
+ }
+ //pub.setContentType(contentType);
+ return pub;
+ }
+
+ /**
+ * Create a publisher that will contain send methods that return
+ * response object to user.
+ * @param producerFilePath set all properties for publishing message
+ * @return MRBatchingPublisher obj
+ * @throws FileNotFoundException exc
+ * @throws IOException ioex
+ */
+ public static MRBatchingPublisher createBatchingPublisher ( final String producerFilePath, boolean withResponse ) throws FileNotFoundException,IOException {
+ FileReader reader = new FileReader(new File (producerFilePath));
+ Properties props = new Properties();
+ props.load(reader);
+ MRSimplerBatchPublisher pub = new MRSimplerBatchPublisher.Builder ().
+ againstUrls(MRConsumerImpl.stringToList(props.getProperty("host"))).
+ onTopic ( props.getProperty("topic") ).
+ batchTo (Integer.parseInt(props.getProperty("maxBatchSize")), Integer.parseInt(props.getProperty("maxAgeMs").toString())).
+ compress (Boolean.parseBoolean(props.getProperty("compress"))).
+ httpThreadTime(Integer.parseInt(props.getProperty("MessageSentThreadOccurance"))).
+ withResponse(withResponse).
+ build ();
+ pub.setHost(props.getProperty("host"));
+ if(props.getProperty("TransportType").equalsIgnoreCase(ProtocolTypeConstants.AUTH_KEY.getValue())){
+
+ pub.setAuthKey(props.getProperty("authKey"));
+ pub.setAuthDate(props.getProperty("authDate"));
+ pub.setUsername(props.getProperty("username"));
+ pub.setPassword(props.getProperty("password"));
+ }else{
+ pub.setUsername(props.getProperty("username"));
+ pub.setPassword(props.getProperty("password"));
+ }
+ pub.setProducerFilePath(producerFilePath);
+ pub.setProtocolFlag(props.getProperty("TransportType"));
+ pub.setProps(props);
+ routeFilePath=props.getProperty("DME2preferredRouterFilePath");
+ routeReader= new FileReader(new File (routeFilePath));
+ prop= new Properties();
+ File fo= new File(routeFilePath);
+ if(!fo.exists()){
+ routeWriter=new FileWriter(new File(routeFilePath));
+ }
+ //pub.setContentType(contentType);
+ return pub;
+ }
+
+
+
+
+
+
+
+
+
+
+
+ /**
+ * Create an identity manager client to work with API keys.
+ * @param hostSet A set of hosts to be used in the URL to MR. Can be "host:port". Use multiple entries to enable failover.
+ * @param apiKey Your API key
+ * @param apiSecret Your API secret
+ * @return an identity manager
+ */
+ public static MRIdentityManager createIdentityManager ( Collection<String> hostSet, String apiKey, String apiSecret )
+ {
+ MRIdentityManager cim;
+ try {
+ cim = new MRMetaClient ( hostSet );
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ cim.setApiCredentials ( apiKey, apiSecret );
+ return cim;
+ }
+
+ /**
+ * Create a topic manager for working with topics.
+ * @param hostSet A set of hosts to be used in the URL to MR. Can be "host:port". Use multiple entries to enable failover.
+ * @param apiKey Your API key
+ * @param apiSecret Your API secret
+ * @return a topic manager
+ */
+ public static MRTopicManager createTopicManager ( Collection<String> hostSet, String apiKey, String apiSecret )
+ {
+ MRMetaClient tmi;
+ try {
+ tmi = new MRMetaClient ( hostSet );
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ tmi.setApiCredentials ( apiKey, apiSecret );
+ return tmi;
+ }
+
+ /**
+ * Inject a consumer. Used to support unit tests.
+ * @param cc
+ */
+ public static void $testInject ( MRConsumer cc )
+ {
+ MRClientBuilders.sfConsumerMock = cc;
+ }
+
+ public static MRConsumer createConsumer(String host, String topic, String username,
+ String password, String group, String id, int i, int j,String protocalFlag,String consumerFilePath) {
+
+ MRConsumerImpl sub;
+ try {
+ sub = new MRConsumerImpl ( MRConsumerImpl.stringToList(host), topic, group, id, i, j, null, null, null );
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ sub.setUsername(username);
+ sub.setPassword(password);
+ sub.setHost(host);
+ sub.setProtocolFlag(protocalFlag);
+ sub.setConsumerFilePath(consumerFilePath);
+ return sub;
+
+ }
+
+ public static MRConsumer createConsumer(String host, String topic, String username,
+ String password, String group, String id,String protocalFlag,String consumerFilePath, int i, int j) {
+
+ MRConsumerImpl sub;
+ try {
+ sub = new MRConsumerImpl ( MRConsumerImpl.stringToList(host), topic, group, id, i, j, null, null, null );
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ sub.setUsername(username);
+ sub.setPassword(password);
+ sub.setHost(host);
+ sub.setProtocolFlag(protocalFlag);
+ sub.setConsumerFilePath(consumerFilePath);
+ return sub;
+
+ }
+
+ public static MRConsumer createConsumer(String consumerFilePath) throws FileNotFoundException,IOException {
+ FileReader reader = new FileReader(new File (consumerFilePath));
+ Properties props = new Properties();
+ props.load(reader);
+ int timeout;
+ if(props.getProperty("timeout")!=null)
+ timeout=Integer.parseInt(props.getProperty("timeout"));
+ else
+ timeout=-1;
+ int limit;
+ if(props.getProperty("limit")!=null)
+ limit=Integer.parseInt(props.getProperty("limit"));
+ else
+ limit=-1;
+ String group;
+ if(props.getProperty("group")==null)
+ group=UUID.randomUUID ().toString();
+ else
+ group=props.getProperty("group");
+ MRConsumerImpl sub=null;
+ if(props.getProperty("TransportType").equalsIgnoreCase(ProtocolTypeConstants.AUTH_KEY.getValue())){
+ sub = new MRConsumerImpl ( MRConsumerImpl.stringToList(props.getProperty("host")), props.getProperty("topic"), group, props.getProperty("id"),timeout, limit, props.getProperty("filter"),props.getProperty("authKey"), props.getProperty("authDate") );
+ sub.setAuthKey(props.getProperty("authKey"));
+ sub.setAuthDate(props.getProperty("authDate"));
+ sub.setUsername(props.getProperty("username"));
+ sub.setPassword(props.getProperty("password"));
+ }else{
+ sub = new MRConsumerImpl ( MRConsumerImpl.stringToList(props.getProperty("host")), props.getProperty("topic"), group, props.getProperty("id"), timeout, limit, props.getProperty("filter"),props.getProperty("username"), props.getProperty("password") );
+ sub.setUsername(props.getProperty("username"));
+ sub.setPassword(props.getProperty("password"));
+ }
+ sub.setRouterFilePath(props.getProperty("DME2preferredRouterFilePath"));
+ sub.setProps(props);
+ sub.setHost(props.getProperty("host"));
+ sub.setProtocolFlag(props.getProperty("TransportType"));
+ //sub.setConsumerFilePath(consumerFilePath);
+ sub.setfFilter(props.getProperty("filter"));
+ routeFilePath=props.getProperty("DME2preferredRouterFilePath");
+ routeReader= new FileReader(new File (routeFilePath));
+ prop= new Properties();
+ File fo= new File(routeFilePath);
+ if(!fo.exists()){
+ routeWriter=new FileWriter(new File(routeFilePath));
+ }
+ return sub;
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/client/MRConsumer.java b/src/main/java/com/att/nsa/mr/client/MRConsumer.java
new file mode 100644
index 0000000..444eb7c
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/MRConsumer.java
@@ -0,0 +1,54 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client;
+
+import java.io.IOException;
+
+import com.att.nsa.mr.client.response.MRConsumerResponse;
+
+public interface MRConsumer extends MRClient
+{
+ /**
+ * Fetch a set of messages. The consumer's timeout and message limit are used if set in the constructor call.
+
+ * @return a set of messages
+ * @throws IOException
+ */
+ Iterable<String> fetch () throws IOException, Exception;
+
+ /**
+ * Fetch a set of messages with an explicit timeout and limit for this call. These values
+ * override any set in the constructor call.
+ *
+ * @param timeoutMs The amount of time in milliseconds that the server should keep the connection
+ * open while waiting for message traffic. Use -1 for default timeout (controlled on the server-side).
+ * @param limit A limit on the number of messages returned in a single call. Use -1 for no limit.
+ * @return a set messages
+ * @throws IOException if there's a problem connecting to the server
+ */
+ Iterable<String> fetch ( int timeoutMs, int limit ) throws IOException, Exception;
+
+ MRConsumerResponse fetchWithReturnConsumerResponse ();
+
+
+ MRConsumerResponse fetchWithReturnConsumerResponse ( int timeoutMs, int limit );
+}
diff --git a/src/main/java/com/att/nsa/mr/client/MRIdentityManager.java b/src/main/java/com/att/nsa/mr/client/MRIdentityManager.java
new file mode 100644
index 0000000..5483761
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/MRIdentityManager.java
@@ -0,0 +1,100 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client;
+
+import java.io.IOException;
+
+import com.att.nsa.apiClient.credentials.ApiCredential;
+import com.att.nsa.apiClient.http.HttpException;
+import com.att.nsa.apiClient.http.HttpObjectNotFoundException;
+
+/**
+ * A client for manipulating API keys.
+ * @author author
+ *
+ */
+public interface MRIdentityManager extends MRClient
+{
+ /**
+ * An API Key record
+ */
+ public interface ApiKey
+ {
+ /**
+ * Get the email address associated with the API key
+ * @return the email address on the API key or null
+ */
+ String getEmail ();
+
+ /**
+ * Get the description associated with the API key
+ * @return the description on the API key or null
+ */
+ String getDescription ();
+ }
+
+ /**
+ * Create a new API key on the UEB cluster. The returned credential instance
+ * contains the new API key and API secret. This is the only time the secret
+ * is available to the client -- there's no API for retrieving it later -- so
+ * your application must store it securely.
+ *
+ * @param email
+ * @param description
+ * @return a new credential
+ * @throws HttpException
+ * @throws MRApiException
+ * @throws IOException
+ */
+ ApiCredential createApiKey ( String email, String description ) throws HttpException, MRApiException, IOException;
+
+ /**
+ * Get basic info about a known API key
+ * @param apiKey
+ * @return the API key's info or null if it doesn't exist
+ * @throws HttpObjectNotFoundException, HttpException, MRApiException
+ * @throws IOException
+ */
+ ApiKey getApiKey ( String apiKey ) throws HttpObjectNotFoundException, HttpException, MRApiException, IOException;
+
+ /**
+ * Update the record for the API key used to authenticate this request. The UEB
+ * API requires that you authenticate with the same key you're updating, so the
+ * API key being changed is the one used for setApiCredentials.
+ *
+ * @param email use null to keep the current value
+ * @param description use null to keep the current value
+ * @throws IOException
+ * @throws HttpException
+ * @throws HttpObjectNotFoundException
+ */
+ void updateCurrentApiKey ( String email, String description ) throws HttpObjectNotFoundException, HttpException, IOException;
+
+ /**
+ * Delete the *current* API key. After this call returns, the API key
+ * used to authenticate will no longer be valid.
+ *
+ * @throws IOException
+ * @throws HttpException
+ */
+ void deleteCurrentApiKey () throws HttpException, IOException;
+}
diff --git a/src/main/java/com/att/nsa/mr/client/MRPublisher.java b/src/main/java/com/att/nsa/mr/client/MRPublisher.java
new file mode 100644
index 0000000..651c548
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/MRPublisher.java
@@ -0,0 +1,93 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client;
+
+import java.io.IOException;
+import java.util.Collection;
+
+/**
+ * A MR publishing interface.
+ *
+ */
+public interface MRPublisher extends MRClient
+{
+ /**
+ * A simple message container
+ */
+ public static class message
+ {
+ public message ( String partition, String msg )
+ {
+ fPartition = partition == null ? "" : partition;
+ fMsg = msg;
+ if ( fMsg == null )
+ {
+ throw new IllegalArgumentException ( "Can't send a null message." );
+ }
+ }
+
+ public message ( message msg )
+ {
+ this ( msg.fPartition, msg.fMsg );
+ }
+
+ public final String fPartition;
+ public final String fMsg;
+ }
+
+ /**
+ * Send the given message without partition. partition will be placed at HTTP request level.
+ * @param msg message to sent
+ * @return the number of pending messages
+ * @throws IOException exception
+ */
+ int send ( String msg ) throws IOException;
+ /**
+ * Send the given message using the given partition.
+ * @param partition partition
+ * @param msg message
+ * @return the number of pending messages
+ * @throws IOException exception
+ */
+ int send ( String partition, String msg ) throws IOException;
+
+ /**
+ * Send the given message using its partition.
+ * @param msg mesg
+ * @return the number of pending messages
+ * @throws IOException exp
+ */
+ int send ( message msg ) throws IOException;
+
+ /**
+ * Send the given messages using their partitions.
+ * @param msgs msg
+ * @return the number of pending messages
+ * @throws IOException exp
+ */
+ int send ( Collection<message> msgs ) throws IOException;
+
+ /**
+ * Close this publisher. It's an error to call send() after close()
+ */
+ void close ();
+}
diff --git a/src/main/java/com/att/nsa/mr/client/MRTopicManager.java b/src/main/java/com/att/nsa/mr/client/MRTopicManager.java
new file mode 100644
index 0000000..13524bd
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/MRTopicManager.java
@@ -0,0 +1,183 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client;
+
+import java.io.IOException;
+import java.util.Set;
+
+import com.att.nsa.apiClient.http.HttpException;
+import com.att.nsa.apiClient.http.HttpObjectNotFoundException;
+
+
+/**
+ * A client for working with topic metadata.
+ * @author author
+ */
+public interface MRTopicManager extends MRClient
+{
+ /**
+ * Get the topics available in the cluster
+ * @return a set of topic names
+ * @throws IOException
+ */
+ Set<String> getTopics () throws IOException;
+
+ /**
+ * Information about a topic.
+ */
+ public interface TopicInfo
+ {
+ /**
+ * Get the owner of the topic
+ * @return the owner, or null if no entry
+ */
+ String getOwner ();
+
+ /**
+ * Get the description for this topic
+ * @return the description, or null if no entry
+ */
+ String getDescription ();
+
+ /**
+ * Get the set of allowed producers (as API keys) on this topic
+ * @return the set of allowed producers, null of no ACL exists/enabled
+ */
+ Set<String> getAllowedProducers ();
+
+ /**
+ * Get the set of allowed consumers (as API keys) on this topic
+ * @return the set of allowed consumers, null of no ACL exists/enabled
+ */
+ Set<String> getAllowedConsumers ();
+ }
+
+ /**
+ * Get information about a topic.
+ * @param topic
+ * @return topic information
+ * @throws IOException
+ * @throws HttpObjectNotFoundException
+ */
+ TopicInfo getTopicMetadata ( String topic ) throws HttpObjectNotFoundException, IOException;
+
+ /**
+ * Create a new topic.
+ * @param topicName
+ * @param topicDescription
+ * @param partitionCount
+ * @param replicationCount
+ * @throws HttpException
+ * @throws IOException
+ */
+ void createTopic ( String topicName, String topicDescription, int partitionCount, int replicationCount ) throws HttpException, IOException;
+
+ /**
+ * Delete the topic. This call must be authenticated and the API key listed as owner on the topic.
+ * NOTE: The MR (UEB) API server does not support topic deletion at this time (mid 2015)
+ * @param topic
+ * @throws HttpException
+ * @throws IOException
+ * @deprecated If/when the Kafka system supports topic delete, or the implementation changes, this will be restored.
+ */
+ @Deprecated
+ void deleteTopic ( String topic ) throws HttpException, IOException;
+
+ /**
+ * Can any client produce events into this topic without authentication?
+ * @param topic
+ * @return true if the topic is open for producing
+ * @throws IOException
+ * @throws HttpObjectNotFoundException
+ */
+ boolean isOpenForProducing ( String topic ) throws HttpObjectNotFoundException, IOException;
+
+ /**
+ * Get the set of allowed producers. If the topic is open, the result is null.
+ * @param topic
+ * @return a set of allowed producers or null
+ * @throws IOException
+ * @throws HttpObjectNotFoundException
+ */
+ Set<String> getAllowedProducers ( String topic ) throws HttpObjectNotFoundException, IOException;
+
+ /**
+ * Allow the given API key to produce messages on the given topic. The caller must
+ * own this topic.
+ * @param topic
+ * @param apiKey
+ * @throws HttpException
+ * @throws HttpObjectNotFoundException
+ * @throws IOException
+ */
+ void allowProducer ( String topic, String apiKey ) throws HttpObjectNotFoundException, HttpException, IOException;
+
+ /**
+ * Revoke the given API key's authorization to produce messages on the given topic.
+ * The caller must own this topic.
+ * @param topic
+ * @param apiKey
+ * @throws HttpException
+ * @throws IOException
+ */
+ void revokeProducer ( String topic, String apiKey ) throws HttpException, IOException;
+
+ /**
+ * Can any client consume events from this topic without authentication?
+ * @param topic
+ * @return true if the topic is open for consuming
+ * @throws IOException
+ * @throws HttpObjectNotFoundException
+ */
+ boolean isOpenForConsuming ( String topic ) throws HttpObjectNotFoundException, IOException;
+
+ /**
+ * Get the set of allowed consumers. If the topic is open, the result is null.
+ * @param topic
+ * @return a set of allowed consumers or null
+ * @throws IOException
+ * @throws HttpObjectNotFoundException
+ */
+ Set<String> getAllowedConsumers ( String topic ) throws HttpObjectNotFoundException, IOException;
+
+ /**
+ * Allow the given API key to consume messages on the given topic. The caller must
+ * own this topic.
+ * @param topic
+ * @param apiKey
+ * @throws HttpException
+ * @throws HttpObjectNotFoundException
+ * @throws IOException
+ */
+ void allowConsumer ( String topic, String apiKey ) throws HttpObjectNotFoundException, HttpException, IOException;
+
+ /**
+ * Revoke the given API key's authorization to consume messages on the given topic.
+ * The caller must own this topic.
+ * @param topic
+ * @param apiKey
+ * @throws HttpException
+ * @throws IOException
+ */
+ void revokeConsumer ( String topic, String apiKey ) throws HttpException, IOException;
+}
+
diff --git a/src/main/java/com/att/nsa/mr/client/impl/Clock.java b/src/main/java/com/att/nsa/mr/client/impl/Clock.java
new file mode 100644
index 0000000..ace791e
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/Clock.java
@@ -0,0 +1,63 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client.impl;
+
+public class Clock
+{
+ public synchronized static Clock getIt ()
+ {
+ if ( sfClock == null )
+ {
+ sfClock = new Clock ();
+ }
+ return sfClock;
+ }
+
+ /**
+ * Get the system's current time in milliseconds.
+ * @return the current time
+ */
+ public static long now ()
+ {
+ return getIt().nowImpl ();
+ }
+
+ /**
+ * Get current time in milliseconds
+ * @return current time in ms
+ */
+ protected long nowImpl ()
+ {
+ return System.currentTimeMillis ();
+ }
+
+ protected Clock ()
+ {
+ }
+
+ private static Clock sfClock = null;
+
+ protected synchronized static void register ( Clock testClock )
+ {
+ sfClock = testClock;
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRBaseClient.java b/src/main/java/com/att/nsa/mr/client/impl/MRBaseClient.java
new file mode 100644
index 0000000..a9fd9bd
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRBaseClient.java
@@ -0,0 +1,393 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client.impl;
+
+import java.net.MalformedURLException;
+import java.util.Collection;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
+
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.Response;
+
+import org.apache.http.HttpException;
+import org.glassfish.jersey.client.authentication.HttpAuthenticationFeature;
+import org.glassfish.jersey.internal.util.Base64;
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.json.JSONTokener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.nsa.apiClient.http.CacheUse;
+import com.att.nsa.apiClient.http.HttpClient;
+import com.att.nsa.mr.client.MRClient;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+//import com.fasterxml.jackson.core.JsonProcessingException;
+
+public class MRBaseClient extends HttpClient implements MRClient
+{
+
+ private static final String MR_AUTH_CONSTANT = "X-CambriaAuth";
+ private static final String MR_DATE_CONSTANT = "X-CambriaDate";
+
+ protected MRBaseClient ( Collection<String> hosts ) throws MalformedURLException
+ {
+ super ( ConnectionType.HTTP, hosts, MRConstants.kStdMRServicePort );
+ }
+
+ protected MRBaseClient ( Collection<String> hosts, int stdSvcPort ) throws MalformedURLException {
+ super ( ConnectionType.HTTP,hosts, stdSvcPort);
+
+ fLog = LoggerFactory.getLogger ( this.getClass().getName () );
+ }
+
+ protected MRBaseClient ( Collection<String> hosts, String clientSignature ) throws MalformedURLException
+ {
+ super(ConnectionType.HTTP, hosts, MRConstants.kStdMRServicePort, clientSignature, CacheUse.NONE, 1, 1L, TimeUnit.MILLISECONDS, 32, 32, 600000);
+
+ fLog = LoggerFactory.getLogger ( this.getClass().getName () );
+ }
+
+
+ @Override
+ public void close ()
+ {
+ }
+
+ protected Set<String> jsonArrayToSet ( JSONArray a )
+ {
+ if ( a == null ) return null;
+
+ final TreeSet<String> set = new TreeSet<String> ();
+ for ( int i=0; i<a.length (); i++ )
+ {
+ set.add ( a.getString ( i ));
+ }
+ return set;
+ }
+
+ public void logTo ( Logger log )
+ {
+ fLog = log;
+ replaceLogger ( log );
+ }
+
+ protected Logger getLog ()
+ {
+ return fLog;
+ }
+
+ private Logger fLog;
+
+ public JSONObject post(final String path, final byte[] data, final String contentType, final String username, final String password, final String protocolFlag) throws HttpException, JSONException{
+ if ((null != username && null != password)) {
+ WebTarget target = null;
+
+ Response response = null;
+
+ target = getTarget(path, username, password);
+ String encoding = Base64.encodeAsString(username+":"+password);
+
+
+ response = target.request().header("Authorization", "Basic " + encoding).post(Entity.entity(data, contentType));
+
+ return getResponseDataInJson(response);
+ } else {
+ throw new HttpException("Authentication Failed: Username/password/AuthKey/AuthDate parameter(s) cannot be null or empty.");
+ }
+ }
+ public String postWithResponse(final String path, final byte[] data, final String contentType, final String username, final String password, final String protocolFlag) throws HttpException, JSONException{
+ String responseData = null;
+ if ((null != username && null != password)) {
+ WebTarget target = null;
+
+ Response response = null;
+
+ target = getTarget(path, username, password);
+ String encoding = Base64.encodeAsString(username+":"+password);
+
+
+ response = target.request().header("Authorization", "Basic " + encoding).post(Entity.entity(data, contentType));
+
+ responseData = response.readEntity(String.class);
+ return responseData;
+ } else {
+ throw new HttpException("Authentication Failed: Username/password/AuthKey/AuthDate parameter(s) cannot be null or empty.");
+ }
+ }
+ public JSONObject postAuth(final String path, final byte[] data, final String contentType, final String authKey,final String authDate,final String username, final String password, final String protocolFlag) throws HttpException, JSONException{
+ if ((null != username && null != password)) {
+ WebTarget target = null;
+
+ Response response = null;
+ target= getTarget(path,username, password);
+ response = target.request()
+ .header(MR_AUTH_CONSTANT, authKey)
+ .header(MR_DATE_CONSTANT, authDate)
+ .post(Entity.entity(data, contentType));
+
+ return getResponseDataInJson(response);
+ } else {
+ throw new HttpException("Authentication Failed: Username/password/AuthKey/AuthDate parameter(s) cannot be null or empty.");
+ }
+ }
+ public String postAuthwithResponse(final String path, final byte[] data, final String contentType, final String authKey,final String authDate,final String username, final String password, final String protocolFlag) throws HttpException, JSONException{
+ String responseData = null;
+ if ((null != username && null != password)) {
+ WebTarget target = null;
+
+ Response response = null;
+ target= getTarget(path,username, password);
+ response = target.request()
+ .header(MR_AUTH_CONSTANT, authKey)
+ .header(MR_DATE_CONSTANT, authDate)
+ .post(Entity.entity(data, contentType));
+ responseData = response.readEntity(String.class);
+ return responseData;
+
+ } else {
+ throw new HttpException("Authentication Failed: Username/password/AuthKey/AuthDate parameter(s) cannot be null or empty.");
+ }
+ }
+
+
+ public JSONObject get(final String path, final String username, final String password, final String protocolFlag) throws HttpException, JSONException {
+ if (null != username && null != password) {
+
+ WebTarget target = null;
+
+ Response response = null;
+ if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
+ target=getTarget(path);
+ response = target.request()
+ .header(MR_AUTH_CONSTANT, username)
+ .header(MR_DATE_CONSTANT, password)
+ .get();
+ } else {
+ target = getTarget(path, username, password);
+ String encoding = Base64.encodeAsString(username+":"+password);
+
+ response = target.request().header("Authorization", "Basic " + encoding).get();
+
+ }
+ return getResponseDataInJson(response);
+ } else {
+ throw new HttpException("Authentication Failed: Username/password/AuthKey/Authdate parameter(s) cannot be null or empty.");
+ }
+ }
+
+
+ public String getResponse(final String path, final String username, final String password, final String protocolFlag) throws HttpException, JSONException {
+ String responseData = null;
+ if (null != username && null != password) {
+
+ WebTarget target = null;
+
+ Response response = null;
+ if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
+ target=getTarget(path);
+ response = target.request()
+ .header(MR_AUTH_CONSTANT, username)
+ .header(MR_DATE_CONSTANT, password)
+ .get();
+ } else {
+ target = getTarget(path, username, password);
+ String encoding = Base64.encodeAsString(username+":"+password);
+ response = target.request().header("Authorization", "Basic " + encoding).get();
+ }
+ MRClientFactory.HTTPHeadersMap=response.getHeaders();
+
+ String transactionid=response.getHeaderString("transactionid");
+ if (transactionid!=null && !transactionid.equalsIgnoreCase("")) {
+ fLog.info("TransactionId : " + transactionid);
+ }
+
+ responseData = response.readEntity(String.class);
+ return responseData;
+ } else {
+ throw new HttpException("Authentication Failed: Username/password/AuthKey/Authdate parameter(s) cannot be null or empty.");
+ }
+ }
+
+ public JSONObject getAuth(final String path, final String authKey, final String authDate,final String username, final String password, final String protocolFlag) throws HttpException, JSONException {
+ if (null != username && null != password) {
+
+ WebTarget target = null;
+
+ Response response = null;
+ target=getTarget(path, username, password);
+ response = target.request()
+ .header(MR_AUTH_CONSTANT, authKey)
+ .header(MR_DATE_CONSTANT, authDate)
+ .get();
+
+ return getResponseDataInJson(response);
+ } else {
+ throw new HttpException("Authentication Failed: Username/password/AuthKey/Authdate parameter(s) cannot be null or empty.");
+ }
+ }
+
+ public String getAuthResponse(final String path, final String authKey, final String authDate,final String username, final String password, final String protocolFlag) throws HttpException, JSONException {
+ String responseData = null;
+ if (null != username && null != password) {
+
+ WebTarget target = null;
+
+ Response response = null;
+ target=getTarget(path, username, password);
+ response = target.request()
+ .header(MR_AUTH_CONSTANT, authKey)
+ .header(MR_DATE_CONSTANT, authDate)
+ .get();
+
+ MRClientFactory.HTTPHeadersMap=response.getHeaders();
+
+ String transactionid=response.getHeaderString("transactionid");
+ if (transactionid!=null && !transactionid.equalsIgnoreCase("")) {
+ fLog.info("TransactionId : " + transactionid);
+ }
+
+ responseData = response.readEntity(String.class);
+ return responseData;
+ } else {
+ throw new HttpException("Authentication Failed: Username/password/AuthKey/Authdate parameter(s) cannot be null or empty.");
+ }
+ }
+
+ private WebTarget getTarget(final String path, final String username, final String password) {
+
+ Client client = ClientBuilder.newClient();
+
+
+ // Using UNIVERSAL as it supports both BASIC and DIGEST authentication types.
+ HttpAuthenticationFeature feature = HttpAuthenticationFeature.universal(username, password);
+ client.register(feature);
+
+ return client.target(path);
+ }
+
+
+ private WebTarget getTarget(final String path) {
+
+ Client client = ClientBuilder.newClient();
+ return client.target(path);
+ }
+ private JSONObject getResponseDataInJson(Response response) throws JSONException {
+ try {
+ MRClientFactory.HTTPHeadersMap=response.getHeaders();
+ // fLog.info("DMAAP response status: " + response.getStatus());
+
+
+ //MultivaluedMap<String, Object> headersMap = response.getHeaders();
+ //for(String key : headersMap.keySet()) {
+ String transactionid=response.getHeaderString("transactionid");
+ if (transactionid!=null && !transactionid.equalsIgnoreCase("")) {
+ fLog.info("TransactionId : " + transactionid);
+ }
+
+ /*final String responseData = response.readEntity(String.class);
+ JSONTokener jsonTokener = new JSONTokener(responseData);
+ JSONObject jsonObject = null;
+ final char firstChar = jsonTokener.next();
+ jsonTokener.back();
+ if ('[' == firstChar) {
+ JSONArray jsonArray = new JSONArray(jsonTokener);
+ jsonObject = new JSONObject();
+ jsonObject.put("result", jsonArray);
+ } else {
+ jsonObject = new JSONObject(jsonTokener);
+ }
+
+ return jsonObject;*/
+
+
+ if(response.getStatus()==403) {
+ JSONObject jsonObject = null;
+ jsonObject = new JSONObject();
+ JSONArray jsonArray = new JSONArray();
+ jsonArray.put(response.getEntity());
+ jsonObject.put("result", jsonArray);
+ jsonObject.put("status", response.getStatus());
+ return jsonObject;
+ }
+ String responseData = response.readEntity(String.class);
+
+ JSONTokener jsonTokener = new JSONTokener(responseData);
+ JSONObject jsonObject = null;
+ final char firstChar = jsonTokener.next();
+ jsonTokener.back();
+ if ('[' == firstChar) {
+ JSONArray jsonArray = new JSONArray(jsonTokener);
+ jsonObject = new JSONObject();
+ jsonObject.put("result", jsonArray);
+ jsonObject.put("status", response.getStatus());
+ } else {
+ jsonObject = new JSONObject(jsonTokener);
+ jsonObject.put("status", response.getStatus());
+ }
+
+ return jsonObject;
+ } catch (JSONException excp) {
+ fLog.error("DMAAP - Error reading response data.", excp);
+ return null;
+ }
+
+ }
+
+ public String getHTTPErrorResponseMessage(String responseString){
+
+ String response = null;
+ int beginIndex = 0;
+ int endIndex = 0;
+ if(responseString.contains("<body>")){
+
+ beginIndex = responseString.indexOf("body>")+5;
+ endIndex = responseString.indexOf("</body");
+ response = responseString.substring(beginIndex,endIndex);
+ }
+
+ return response;
+
+ }
+
+ public String getHTTPErrorResponseCode(String responseString){
+
+ String response = null;
+ int beginIndex = 0;
+ int endIndex = 0;
+ if(responseString.contains("<title>")){
+ beginIndex = responseString.indexOf("title>")+6;
+ endIndex = responseString.indexOf("</title");
+ response = responseString.substring(beginIndex,endIndex);
+ }
+
+ return response;
+ }
+
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRBatchPublisher.java b/src/main/java/com/att/nsa/mr/client/impl/MRBatchPublisher.java
new file mode 100644
index 0000000..597e4da
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRBatchPublisher.java
@@ -0,0 +1,485 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client.impl;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.MalformedURLException;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
+import java.util.zip.GZIPOutputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.nsa.apiClient.http.HttpClient;
+import com.att.nsa.apiClient.http.HttpException;
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.response.MRPublisherResponse;
+
+/**
+ * This is a batching publisher class that allows the client to publish messages
+ * in batches that are limited in terms of size and/or hold time.
+ *
+ * @author author
+ * @deprecated This class's tricky locking doesn't quite work
+ *
+ */
+@Deprecated
+public class MRBatchPublisher implements MRBatchingPublisher
+{
+ public static final long kMinMaxAgeMs = 1;
+
+ /**
+ * Create a batch publisher.
+ *
+ * @param baseUrls the base URLs, like "localhost:8080". This class adds the correct application path.
+ * @param topic the topic to publish to
+ * @param maxBatchSize the maximum size of a batch
+ * @param maxAgeMs the maximum age of a batch
+ */
+ public MRBatchPublisher ( Collection<String> baseUrls, String topic, int maxBatchSize, long maxAgeMs, boolean compress )
+ {
+ if ( maxAgeMs < kMinMaxAgeMs )
+ {
+ fLog.warn ( "Max age in ms is less than the minimum. Overriding to " + kMinMaxAgeMs );
+ maxAgeMs = kMinMaxAgeMs;
+ }
+
+ try {
+ fSender = new Sender ( baseUrls, topic, maxBatchSize, maxAgeMs, compress );
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+
+ // FIXME: this strategy needs an overhaul -- why not just run a thread that knows how to wait for
+ // the oldest msg to hit max age? (locking is complicated, but should be do-able)
+ fExec = new ScheduledThreadPoolExecutor ( 1 );
+ fExec.scheduleAtFixedRate ( fSender, 100, 50, TimeUnit.MILLISECONDS );
+ }
+
+ @Override
+ public void setApiCredentials ( String apiKey, String apiSecret )
+ {
+ fSender.setApiCredentials ( apiKey, apiSecret );
+ }
+
+ @Override
+ public void clearApiCredentials ()
+ {
+ fSender.clearApiCredentials ();
+ }
+
+ /**
+ * Send the given message with the given partition
+ * @param partition
+ * @param msg
+ * @throws IOException
+ */
+ @Override
+ public int send ( String partition, String msg ) throws IOException
+ {
+ return send ( new message ( partition, msg ) );
+ }
+ @Override
+ public int send ( String msg ) throws IOException
+ {
+ return send ( new message ( "",msg ) );
+ }
+ /**
+ * Send the given message
+ * @param userMsg a message
+ * @throws IOException
+ */
+ @Override
+ public int send ( message userMsg ) throws IOException
+ {
+ final LinkedList<message> list = new LinkedList<message> ();
+ list.add ( userMsg );
+ return send ( list );
+ }
+
+ /**
+ * Send the given set of messages
+ * @param msgs the set of messages, sent in order of iteration
+ * @return the number of messages in the pending queue (this could actually be less than the size of the given collection, depending on thread timing)
+ * @throws IOException
+ */
+ @Override
+ public int send ( Collection<message> msgs ) throws IOException
+ {
+ if ( msgs.size() > 0 )
+ {
+ fSender.queue ( msgs );
+ }
+ return fSender.size ();
+ }
+
+ @Override
+ public int getPendingMessageCount ()
+ {
+ return fSender.size ();
+ }
+
+ /**
+ * Send any pending messages and close this publisher.
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ @Override
+ public void close ()
+ {
+ try
+ {
+ final List<message> remains = close ( Long.MAX_VALUE, TimeUnit.MILLISECONDS );
+ if ( remains.size() > 0 )
+ {
+ fLog.warn ( "Closing publisher with " + remains.size() + " messages unsent. "
+ + "(Consider using the alternate close method to capture unsent messages in this case.)" );
+ }
+ }
+ catch ( InterruptedException e )
+ {
+ fLog.warn ( "Possible message loss. " + e.getMessage(), e );
+ }
+ catch ( IOException e )
+ {
+ fLog.warn ( "Possible message loss. " + e.getMessage(), e );
+ }
+ }
+
+ public List<message> close ( long time, TimeUnit unit ) throws InterruptedException, IOException
+ {
+ fExec.setContinueExistingPeriodicTasksAfterShutdownPolicy ( false );
+ fExec.setExecuteExistingDelayedTasksAfterShutdownPolicy ( false );
+ fExec.shutdown ();
+
+ final long waitInMs = TimeUnit.MILLISECONDS.convert ( time, unit );
+ final long timeoutAtMs = System.currentTimeMillis () + waitInMs;
+ while ( System.currentTimeMillis () < timeoutAtMs && getPendingMessageCount() > 0 )
+ {
+ fSender.checkSend ( true );
+ Thread.sleep ( 250 );
+ }
+
+ final LinkedList<message> result = new LinkedList<message> ();
+ fSender.drainTo ( result );
+ return result;
+ }
+
+ private final ScheduledThreadPoolExecutor fExec;
+ private final Sender fSender;
+
+ private static class TimestampedMessage extends message
+ {
+ public TimestampedMessage ( message m )
+ {
+ super ( m );
+ timestamp = System.currentTimeMillis ();
+ }
+ public final long timestamp;
+ }
+
+ private Logger fLog = LoggerFactory.getLogger ( MRBatchPublisher.class );
+
+ private class Sender extends MRBaseClient implements Runnable
+ {
+ public Sender ( Collection<String> baseUrls, String topic, int maxBatch, long maxAgeMs, boolean compress ) throws MalformedURLException
+ {
+ super ( baseUrls );
+
+ fNextBatch = new LinkedList<TimestampedMessage> ();
+ fSendingBatch = null;
+ fTopic = topic;
+ fMaxBatchSize = maxBatch;
+ fMaxAgeMs = maxAgeMs;
+ fCompress = compress;
+ fLock = new ReentrantReadWriteLock ();
+ fWriteLock = fLock.writeLock ();
+ fReadLock = fLock.readLock ();
+ fDontSendUntilMs = 0;
+ }
+
+ public void drainTo ( LinkedList<message> list )
+ {
+ fWriteLock.lock ();
+ try
+ {
+ if ( fSendingBatch != null )
+ {
+ list.addAll ( fSendingBatch );
+ }
+ list.addAll ( fNextBatch );
+
+ fSendingBatch = null;
+ fNextBatch.clear ();
+ }
+ finally
+ {
+ fWriteLock.unlock ();
+ }
+ }
+
+ /**
+ * Called periodically by the background executor.
+ */
+ @Override
+ public void run ()
+ {
+ try
+ {
+ checkSend ( false );
+ }
+ catch ( IOException e )
+ {
+ fLog.warn ( "MR background send: " + e.getMessage () );
+ }
+ }
+
+ public int size ()
+ {
+ fReadLock.lock ();
+ try
+ {
+ return fNextBatch.size () + ( fSendingBatch == null ? 0 : fSendingBatch.size () );
+ }
+ finally
+ {
+ fReadLock.unlock ();
+ }
+ }
+
+ /**
+ * Called to queue a message.
+ * @param m
+ * @throws IOException
+ */
+ public void queue ( Collection<message> msgs ) throws IOException
+ {
+ fWriteLock.lock ();
+ try
+ {
+ for ( message userMsg : msgs )
+ {
+ if ( userMsg != null )
+ {
+ fNextBatch.add ( new TimestampedMessage ( userMsg ) );
+ }
+ else
+ {
+ fLog.warn ( "MRBatchPublisher::Sender::queue received a null message." );
+ }
+ }
+ }
+ finally
+ {
+ fWriteLock.unlock();
+ }
+ checkSend ( false );
+ }
+
+ /**
+ * Send a batch if the queue is long enough, or the first pending message is old enough.
+ * @param force
+ * @throws IOException
+ */
+ public void checkSend ( boolean force ) throws IOException
+ {
+ // hold a read lock just long enough to evaluate whether a batch
+ // should be sent
+ boolean shouldSend = false;
+ fReadLock.lock ();
+ try
+ {
+ if ( fNextBatch.size () > 0 )
+ {
+ final long nowMs = System.currentTimeMillis ();
+ shouldSend = ( force || fNextBatch.size() >= fMaxBatchSize );
+ if ( !shouldSend )
+ {
+ final long sendAtMs = fNextBatch.getFirst ().timestamp + fMaxAgeMs;
+ shouldSend = sendAtMs <= nowMs;
+ }
+
+ // however, unless forced, wait after an error
+ shouldSend = force || ( shouldSend && nowMs >= fDontSendUntilMs );
+ }
+ // else: even in 'force', there's nothing to send, so shouldSend=false is fine
+ }
+ finally
+ {
+ fReadLock.unlock ();
+ }
+
+ // if a send is required, acquire a write lock, swap out the next batch,
+ // swap in a fresh batch, and release the lock for the caller to start
+ // filling a batch again. After releasing the lock, send the current
+ // batch. (There could be more messages added between read unlock and
+ // write lock, but that's fine.)
+ if ( shouldSend )
+ {
+ fSendingBatch = null;
+
+ fWriteLock.lock ();
+ try
+ {
+ fSendingBatch = fNextBatch;
+ fNextBatch = new LinkedList<TimestampedMessage> ();
+ }
+ finally
+ {
+ fWriteLock.unlock ();
+ }
+
+ if ( !doSend ( fSendingBatch, this, fTopic, fCompress, fLog ) )
+ {
+ fLog.warn ( "Send failed, rebuilding send queue." );
+
+ // note the time for back-off
+ fDontSendUntilMs = sfWaitAfterError + System.currentTimeMillis ();
+
+ // the send failed. reconstruct the pending queue
+ fWriteLock.lock ();
+ try
+ {
+ final LinkedList<TimestampedMessage> nextGroup = fNextBatch;
+ fNextBatch = fSendingBatch;
+ fNextBatch.addAll ( nextGroup );
+ fSendingBatch = null;
+ fLog.info ( "Send queue rebuilt; " + fNextBatch.size () + " messages to send." );
+ }
+ finally
+ {
+ fWriteLock.unlock ();
+ }
+ }
+ else
+ {
+ fWriteLock.lock ();
+ try
+ {
+ fSendingBatch = null;
+ }
+ finally
+ {
+ fWriteLock.unlock ();
+ }
+ }
+ }
+ }
+
+ private LinkedList<TimestampedMessage> fNextBatch;
+ private LinkedList<TimestampedMessage> fSendingBatch;
+ private final String fTopic;
+ private final int fMaxBatchSize;
+ private final long fMaxAgeMs;
+ private final boolean fCompress;
+ private final ReentrantReadWriteLock fLock;
+ private final WriteLock fWriteLock;
+ private final ReadLock fReadLock;
+ private long fDontSendUntilMs;
+ private static final long sfWaitAfterError = 1000;
+ }
+
+ // this is static so that it's clearly not using any mutable member data outside of a lock
+ private static boolean doSend ( LinkedList<TimestampedMessage> toSend, HttpClient client, String topic, boolean compress, Logger log )
+ {
+ // it's possible for this call to be made with an empty list. in this case, just return.
+ if ( toSend.size() < 1 )
+ {
+ return true;
+ }
+
+ final long nowMs = System.currentTimeMillis ();
+ final String url = MRConstants.makeUrl ( topic );
+
+ log.info ( "sending " + toSend.size() + " msgs to " + url + ". Oldest: " + ( nowMs - toSend.getFirst().timestamp ) + " ms" );
+
+ final ByteArrayOutputStream baseStream = new ByteArrayOutputStream ();
+ try
+ {
+ OutputStream os = baseStream;
+ if ( compress )
+ {
+ os = new GZIPOutputStream ( baseStream );
+ }
+ for ( TimestampedMessage m : toSend )
+ {
+ os.write ( ( "" + m.fPartition.length () ).getBytes() );
+ os.write ( '.' );
+ os.write ( ( "" + m.fMsg.length () ).getBytes() );
+ os.write ( '.' );
+ os.write ( m.fPartition.getBytes() );
+ os.write ( m.fMsg.getBytes() );
+ os.write ( '\n' );
+ }
+ os.close ();
+ }
+ catch ( IOException e )
+ {
+ log.warn ( "Problem writing stream to post: " + e.getMessage () );
+ return false;
+ }
+
+ boolean result = false;
+ final long startMs = System.currentTimeMillis ();
+ try
+ {
+ client.post ( url, compress ?
+ MRFormat.CAMBRIA_ZIP.toString () :
+ MRFormat.CAMBRIA.toString (),
+ baseStream.toByteArray(), false );
+ result = true;
+ }
+ catch ( HttpException e )
+ {
+ log.warn ( "Problem posting to MR: " + e.getMessage() );
+ }
+ catch ( IOException e )
+ {
+ log.warn ( "Problem posting to MR: " + e.getMessage() );
+ }
+
+ log.info ( "MR response (" + (System.currentTimeMillis ()-startMs) + " ms): OK" );
+ return result;
+ }
+
+ @Override
+ public void logTo ( Logger log )
+ {
+ fLog = log;
+ }
+
+ @Override
+ public MRPublisherResponse sendBatchWithResponse() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRClientVersionInfo.java b/src/main/java/com/att/nsa/mr/client/impl/MRClientVersionInfo.java
new file mode 100644
index 0000000..fa9207d
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRClientVersionInfo.java
@@ -0,0 +1,54 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Properties;
+
+public class MRClientVersionInfo
+{
+ public static String getVersion ()
+ {
+ return version;
+ }
+
+ private static final Properties props = new Properties();
+ private static final String version;
+ static
+ {
+ String use = null;
+ try
+ {
+ final InputStream is = MRClientVersionInfo.class.getResourceAsStream ( "/MRClientVersion.properties" );
+ if ( is != null )
+ {
+ props.load ( is );
+ use = props.getProperty ( "MRClientVersion", null );
+ }
+ }
+ catch ( IOException e )
+ {
+ }
+ version = use;
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRConstants.java b/src/main/java/com/att/nsa/mr/client/impl/MRConstants.java
new file mode 100644
index 0000000..803670c
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRConstants.java
@@ -0,0 +1,180 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client.impl;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.http.HttpHost;
+
+class MRConstants
+{
+ private static final String PROTOCOL = "http";
+ public static final String context = "/";
+ public static final String kBasePath = "events/";
+ //public static final int kStdMRServicePort = 3904;
+ public static final int kStdMRServicePort = 8080;
+
+ public static String escape ( String s )
+ {
+ try
+ {
+ return URLEncoder.encode ( s, "UTF-8");
+ }
+ catch ( UnsupportedEncodingException e )
+ {
+ throw new RuntimeException ( e );
+ }
+ }
+
+ public static String makeUrl ( String rawTopic )
+ {
+ final String cleanTopic = escape ( rawTopic );
+
+ final StringBuffer url = new StringBuffer().
+ append ( MRConstants.context ).
+ append ( MRConstants.kBasePath ).
+ append ( cleanTopic );
+ return url.toString ();
+ }
+
+ public static String makeUrl ( final String host, final String rawTopic )
+ {
+ final String cleanTopic = escape ( rawTopic );
+
+ final StringBuffer url = new StringBuffer();
+
+ if (!host.startsWith("http") || !host.startsWith("https") ) {
+ url.append( PROTOCOL + "://" );
+ }
+ url.append(host);
+ url.append ( MRConstants.context );
+ url.append ( MRConstants.kBasePath );
+ url.append ( cleanTopic );
+ return url.toString ();
+ }
+
+ public static String makeUrl ( final String host, final String rawTopic, final String transferprotocol,final String parttion )
+ {
+ final String cleanTopic = escape ( rawTopic );
+
+ final StringBuffer url = new StringBuffer();
+
+ if (transferprotocol !=null && !transferprotocol.equals("")) {
+ url.append( transferprotocol + "://" );
+ }else{
+ url.append( PROTOCOL + "://" );
+ }
+ url.append(host);
+ url.append ( MRConstants.context );
+ url.append ( MRConstants.kBasePath );
+ url.append ( cleanTopic );
+ if(parttion!=null && !parttion.equalsIgnoreCase(""))
+ url.append("?partitionKey=").append(parttion);
+ return url.toString ();
+ }
+ public static String makeConsumerUrl ( String topic, String rawConsumerGroup, String rawConsumerId )
+ {
+ final String cleanConsumerGroup = escape ( rawConsumerGroup );
+ final String cleanConsumerId = escape ( rawConsumerId );
+ return MRConstants.context + MRConstants.kBasePath + topic + "/" + cleanConsumerGroup + "/" + cleanConsumerId;
+ }
+
+ /**
+ * Create a list of HttpHosts from an input list of strings. Input strings have
+ * host[:port] as format. If the port section is not provided, the default port is used.
+ *
+ * @param hosts
+ * @return a list of hosts
+ */
+ public static List<HttpHost> createHostsList(Collection<String> hosts)
+ {
+ final ArrayList<HttpHost> convertedHosts = new ArrayList<HttpHost> ();
+ for ( String host : hosts )
+ {
+ if ( host.length () == 0 ) continue;
+ convertedHosts.add ( hostForString ( host ) );
+ }
+ return convertedHosts;
+ }
+
+ /**
+ * Return an HttpHost from an input string. Input string has
+ * host[:port] as format. If the port section is not provided, the default port is used.
+ *
+ * @param hosts
+ * @return a list of hosts
+ */
+ public static HttpHost hostForString ( String host )
+ {
+ if ( host.length() < 1 ) throw new IllegalArgumentException ( "An empty host entry is invalid." );
+
+ String hostPart = host;
+ int port = kStdMRServicePort;
+
+ final int colon = host.indexOf ( ':' );
+ if ( colon == 0 ) throw new IllegalArgumentException ( "Host entry '" + host + "' is invalid." );
+ if ( colon > 0 )
+ {
+ hostPart = host.substring ( 0, colon ).trim();
+
+ final String portPart = host.substring ( colon + 1 ).trim();
+ if ( portPart.length () > 0 )
+ {
+ try
+ {
+ port = Integer.parseInt ( portPart );
+ }
+ catch ( NumberFormatException x )
+ {
+ throw new IllegalArgumentException ( "Host entry '" + host + "' is invalid.", x );
+ }
+ }
+ // else: use default port on "foo:"
+ }
+
+ return new HttpHost ( hostPart, port );
+ }
+
+ public static String makeConsumerUrl(String host, String fTopic, String fGroup, String fId,final String transferprotocol) {
+ final String cleanConsumerGroup = escape ( fGroup );
+ final String cleanConsumerId = escape ( fId );
+
+ StringBuffer url = new StringBuffer();
+
+ if (transferprotocol !=null && !transferprotocol.equals("")) {
+ url.append( transferprotocol + "://" );
+ }else{
+ url.append( PROTOCOL + "://" );
+ }
+
+ url.append(host);
+ url.append(context);
+ url.append(kBasePath);
+ url.append(fTopic + "/" + cleanConsumerGroup + "/" + cleanConsumerId);
+
+ return url.toString();
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java b/src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java
new file mode 100644
index 0000000..fdd1b89
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java
@@ -0,0 +1,709 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client.impl;
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.http.HttpException;
+import org.apache.http.HttpStatus;
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.json.JSONTokener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.aft.dme2.api.DME2Client;
+import com.att.aft.dme2.api.DME2Exception;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRConsumer;
+import com.att.nsa.mr.client.response.MRConsumerResponse;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+public class MRConsumerImpl extends MRBaseClient implements MRConsumer
+{
+
+ private static final String SUCCESS_MESSAGE = "Success";
+
+
+ private Logger fLog = LoggerFactory.getLogger ( this.getClass().getName () );
+ public static List<String> stringToList ( String str )
+ {
+ final LinkedList<String> set = new LinkedList<String> ();
+ if ( str != null )
+ {
+ final String[] parts = str.trim ().split ( "," );
+ for ( String part : parts )
+ {
+ final String trimmed = part.trim();
+ if ( trimmed.length () > 0 )
+ {
+ set.add ( trimmed );
+ }
+ }
+ }
+ return set;
+ }
+
+ public MRConsumerImpl ( Collection<String> hostPart, final String topic, final String consumerGroup,
+ final String consumerId, int timeoutMs, int limit, String filter, String apiKey_username, String apiSecret_password ) throws MalformedURLException
+ {
+ this( hostPart, topic, consumerGroup, consumerId, timeoutMs, limit, filter, apiKey_username, apiSecret_password, false );
+ }
+
+ public MRConsumerImpl ( Collection<String> hostPart, final String topic, final String consumerGroup,
+ final String consumerId, int timeoutMs, int limit, String filter, String apiKey, String apiSecret, boolean allowSelfSignedCerts ) throws MalformedURLException
+ {
+ super ( hostPart, topic + "::" + consumerGroup + "::" + consumerId );
+
+ fTopic = topic;
+ fGroup = consumerGroup;
+ fId = consumerId;
+ fTimeoutMs = timeoutMs;
+ fLimit = limit;
+ fFilter = filter;
+
+ //setApiCredentials ( apiKey, apiSecret );
+ }
+
+ @Override
+ public Iterable<String> fetch () throws IOException,Exception
+ {
+ // fetch with the timeout and limit set in constructor
+ return fetch ( fTimeoutMs, fLimit );
+ }
+
+ @Override
+ public Iterable<String> fetch ( int timeoutMs, int limit ) throws IOException,Exception
+ {
+ final LinkedList<String> msgs = new LinkedList<String> ();
+
+// FIXME: the timeout on the socket needs to be at least as long as the long poll
+// // sanity check for long poll timeout vs. socket read timeout
+// final int maxReasonableTimeoutMs = CambriaSingletonHttpClient.sfSoTimeoutMs * 9/10;
+// if ( timeoutMs > maxReasonableTimeoutMs )
+// {
+// log.warn ( "Long poll time (" + timeoutMs + ") is too high w.r.t. socket read timeout (" +
+// CambriaSingletonHttpClient.sfSoTimeoutMs + "). Reducing long poll timeout to " + maxReasonableTimeoutMs + "." );
+// timeoutMs = maxReasonableTimeoutMs;
+// }
+
+ // final String urlPath = createUrlPath ( timeoutMs, limit );
+
+ //getLog().info ( "UEB GET " + urlPath );
+ try
+ {
+ if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) {
+ DMEConfigure(timeoutMs, limit);
+ try
+ {
+ //getLog().info ( "Receiving msgs from: " + url+subContextPath );
+ String reply = sender.sendAndWait(timeoutMs+10000L);
+ // System.out.println("Message received = "+reply);
+ final JSONObject o =getResponseDataInJson(reply);
+ //msgs.add(reply);
+ if ( o != null )
+ {
+ final JSONArray a = o.getJSONArray ( "result" );
+ // final int b = o.getInt("status" );
+ //if ( a != null && a.length()>0 )
+ if ( a != null)
+ {
+ for ( int i=0; i<a.length (); i++ )
+ {
+ //msgs.add("DMAAP response status: "+Integer.toString(b));
+ if (a.get(i) instanceof String)
+ msgs.add ( a.getString(i) );
+ else
+ msgs.add ( a.getJSONObject(i).toString() );
+
+
+ }
+ }
+// else if(a != null && a.length()<1){
+// msgs.add ("[]");
+// }
+ }
+ }
+ catch ( JSONException e )
+ {
+ // unexpected response
+ reportProblemWithResponse ();
+ }
+ catch ( HttpException e )
+ {
+ throw new IOException ( e );
+ }
+ }
+
+ if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) {
+ final String urlPath = createUrlPath (MRConstants.makeConsumerUrl ( host, fTopic, fGroup, fId,props.getProperty("Protocol")), timeoutMs, limit );
+
+
+ try
+ {
+ final JSONObject o = get ( urlPath, username, password, protocolFlag );
+
+ if ( o != null )
+ {
+ final JSONArray a = o.getJSONArray ( "result" );
+ final int b = o.getInt("status" );
+ //if ( a != null && a.length()>0 )
+ if ( a != null)
+ {
+ for ( int i=0; i<a.length (); i++ )
+ {
+ msgs.add("DMAAP response status: "+Integer.toString(b));
+ if (a.get(i) instanceof String)
+ msgs.add ( a.getString(i) );
+ else
+ msgs.add ( a.getJSONObject(i).toString() );
+
+ }
+ }
+// else if(a != null && a.length()<1)
+// {
+// msgs.add ("[]");
+// }
+ }
+ }
+ catch ( JSONException e )
+ {
+ // unexpected response
+ reportProblemWithResponse ();
+ }
+ catch ( HttpException e )
+ {
+ throw new IOException ( e );
+ }
+ }
+
+ if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
+ final String urlPath = createUrlPath (MRConstants.makeConsumerUrl ( host, fTopic, fGroup, fId ,props.getProperty("Protocol")), timeoutMs, limit );
+
+
+ try
+ {
+ final JSONObject o = getAuth(urlPath, authKey, authDate, username, password, protocolFlag );
+ if ( o != null )
+ {
+ final JSONArray a = o.getJSONArray ( "result" );
+ final int b = o.getInt("status" );
+ //if ( a != null && a.length()>0)
+ if ( a != null)
+ {
+ for ( int i=0; i<a.length (); i++ )
+ {
+ msgs.add("DMAAP response status: "+Integer.toString(b));
+ if (a.get(i) instanceof String)
+ msgs.add ( a.getString(i) );
+ else
+ msgs.add ( a.getJSONObject(i).toString() );
+
+ }
+ }
+// else if(a != null && a.length()<1){
+// msgs.add ("[]");
+// }
+ }
+ }
+ catch ( JSONException e )
+ {
+ // unexpected response
+ reportProblemWithResponse ();
+ }
+ catch ( HttpException e )
+ {
+ throw new IOException ( e );
+ }
+
+ }
+
+ } catch ( JSONException e ) {
+ // unexpected response
+ reportProblemWithResponse ();
+ } catch (HttpException e) {
+ throw new IOException(e);
+ } catch (Exception e ) {
+ throw e;
+ }
+
+
+ return msgs;
+ }
+
+ private JSONObject getResponseDataInJson(String response) {
+ try {
+
+
+ //fLog.info("DMAAP response status: " + response.getStatus());
+
+ // final String responseData = response.readEntity(String.class);
+ JSONTokener jsonTokener = new JSONTokener(response);
+ JSONObject jsonObject = null;
+ final char firstChar = jsonTokener.next();
+ jsonTokener.back();
+ if ('[' == firstChar) {
+ JSONArray jsonArray = new JSONArray(jsonTokener);
+ jsonObject = new JSONObject();
+ jsonObject.put("result", jsonArray);
+ } else {
+ jsonObject = new JSONObject(jsonTokener);
+ }
+
+ return jsonObject;
+ } catch (JSONException excp) {
+ // fLog.error("DMAAP - Error reading response data.", excp);
+ return null;
+ }
+
+
+
+}
+
+ private JSONObject getResponseDataInJsonWithResponseReturned(String response) {
+ JSONTokener jsonTokener = new JSONTokener(response);
+ JSONObject jsonObject = null;
+ final char firstChar = jsonTokener.next();
+ jsonTokener.back();
+ if(null != response && response.length()==0){
+ return null;
+ }
+
+ if ('[' == firstChar) {
+ JSONArray jsonArray = new JSONArray(jsonTokener);
+ jsonObject = new JSONObject();
+ jsonObject.put("result", jsonArray);
+ } else if('{' == firstChar){
+ return null;
+ } else if('<' == firstChar){
+ return null;
+ }else{
+ jsonObject = new JSONObject(jsonTokener);
+ }
+
+ return jsonObject;
+
+ }
+
+ private final String fTopic;
+ private final String fGroup;
+ private final String fId;
+ private final int fTimeoutMs;
+ private final int fLimit;
+ private String fFilter;
+ private String username;
+ private String password;
+ private String host;
+ private String latitude;
+ private String longitude;
+ private String version;
+ private String serviceName;
+ private String env;
+ private String partner;
+ private String routeOffer;
+ private String subContextPath;
+ private String protocol;
+ private String methodType;
+ private String url;
+ private String dmeuser;
+ private String dmepassword;
+ private String contenttype;
+ private DME2Client sender;
+ public String protocolFlag = ProtocolTypeConstants.DME2.getValue();
+ public String consumerFilePath;
+ private String authKey;
+ private String authDate;
+ private Properties props;
+ private HashMap<String, String> DMETimeOuts;
+ private String handlers;
+ public static String routerFilePath;
+ public static String getRouterFilePath() {
+ return routerFilePath;
+ }
+
+ public static void setRouterFilePath(String routerFilePath) {
+ MRSimplerBatchPublisher.routerFilePath = routerFilePath;
+ }
+ public String getConsumerFilePath() {
+ return consumerFilePath;
+ }
+
+ public void setConsumerFilePath(String consumerFilePath) {
+ this.consumerFilePath = consumerFilePath;
+ }
+
+ public String getProtocolFlag() {
+ return protocolFlag;
+ }
+
+ public void setProtocolFlag(String protocolFlag) {
+ this.protocolFlag = protocolFlag;
+ }
+
+ private void DMEConfigure(int timeoutMs, int limit)throws IOException,DME2Exception, URISyntaxException{
+ latitude = props.getProperty("Latitude");
+ longitude = props.getProperty("Longitude");
+ version = props.getProperty("Version");
+ serviceName = props.getProperty("ServiceName");
+ env = props.getProperty("Environment");
+ partner = props.getProperty("Partner");
+ routeOffer = props.getProperty("routeOffer");
+
+ subContextPath=props.getProperty("SubContextPath")+fTopic+"/"+fGroup+"/"+fId;
+ // subContextPath=createUrlPath (subContextPath, timeoutMs, limit);
+ //if (timeoutMs != -1) subContextPath=createUrlPath (subContextPath, timeoutMs);
+
+ protocol = props.getProperty("Protocol");
+ methodType = props.getProperty("MethodType");
+ dmeuser = props.getProperty("username");
+ dmepassword = props.getProperty("password");
+ contenttype = props.getProperty("contenttype");
+ handlers = props.getProperty("sessionstickinessrequired");
+ //url =protocol+"://DME2SEARCH/"+ "service="+serviceName+"/"+"version="+version+"/"+"envContext="+env+"/"+"partner="+partner;
+ // url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&routeOffer="+partner;
+
+ /**
+ * Changes to DME2Client url to use Partner for auto failover between data centers
+ * When Partner value is not provided use the routeOffer value for auto failover within a cluster
+ */
+
+ String preferredRouteKey = readRoute("preferredRouteKey");
+
+ if (partner != null && !partner.isEmpty() && preferredRouteKey != null && !preferredRouteKey.isEmpty())
+ {
+ url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&partner="+partner+"&routeoffer="+preferredRouteKey;
+ }else if (partner != null && !partner.isEmpty())
+ {
+ url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&partner="+partner;
+ }
+ else if (routeOffer!=null && !routeOffer.isEmpty())
+ {
+ url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&routeoffer="+routeOffer;
+ }
+
+ //fLog.info("url :"+url);
+
+ if(timeoutMs != -1 )url=url+"&timeout="+timeoutMs;
+ if(limit != -1 )url=url+"&limit="+limit;
+
+ DMETimeOuts = new HashMap<String, String>();
+ DMETimeOuts.put("AFT_DME2_EP_READ_TIMEOUT_MS", props.getProperty("AFT_DME2_EP_READ_TIMEOUT_MS"));
+ DMETimeOuts.put("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", props.getProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS"));
+ DMETimeOuts.put("AFT_DME2_EP_CONN_TIMEOUT", props.getProperty("AFT_DME2_EP_CONN_TIMEOUT"));
+ DMETimeOuts.put("Content-Type", contenttype);
+ System.setProperty("AFT_LATITUDE", latitude);
+ System.setProperty("AFT_LONGITUDE", longitude);
+ System.setProperty("AFT_ENVIRONMENT",props.getProperty("AFT_ENVIRONMENT"));
+ // System.setProperty("DME2.DEBUG", "true");
+
+ //SSL changes
+ System.setProperty("AFT_DME2_CLIENT_SSL_INCLUDE_PROTOCOLS",
+ "SSLv3,TLSv1,TLSv1.1");
+ System.setProperty("AFT_DME2_CLIENT_IGNORE_SSL_CONFIG", "false");
+ System.setProperty("AFT_DME2_CLIENT_KEYSTORE_PASSWORD", "changeit");
+ //SSL changes
+
+ sender = new DME2Client(new URI(url), timeoutMs+10000L);
+ sender.setAllowAllHttpReturnCodes(true);
+ sender.setMethod(methodType);
+ sender.setSubContext(subContextPath);
+ if(dmeuser != null && dmepassword != null){
+ sender.setCredentials(dmeuser, dmepassword);
+ //System.out.println(dmepassword);
+ }
+ sender.setHeaders(DMETimeOuts);
+ sender.setPayload("");
+
+ if(handlers.equalsIgnoreCase("yes")){
+ sender.addHeader("AFT_DME2_EXCHANGE_REQUEST_HANDLERS", props.getProperty("AFT_DME2_EXCHANGE_REQUEST_HANDLERS"));
+ sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", props.getProperty("AFT_DME2_EXCHANGE_REPLY_HANDLERS"));
+ sender.addHeader("AFT_DME2_REQ_TRACE_ON", props.getProperty("AFT_DME2_REQ_TRACE_ON"));
+ }else{
+ sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", "com.att.nsa.mr.dme.client.HeaderReplyHandler");
+ }
+ /* HeaderReplyHandler headerhandler= new HeaderReplyHandler();
+ sender.setReplyHandler(headerhandler);*/
+// } catch (DME2Exception x) {
+// getLog().warn(x.getMessage(), x);
+// System.out.println("XXXXXXXXXXXX"+x);
+// } catch (URISyntaxException x) {
+// System.out.println(x);
+// getLog().warn(x.getMessage(), x);
+// } catch (Exception x) {
+// System.out.println("XXXXXXXXXXXX"+x);
+// getLog().warn(x.getMessage(), x);
+// }
+ }
+ public Properties getProps() {
+ return props;
+ }
+
+ public void setProps(Properties props) {
+ this.props = props;
+ }
+
+ protected String createUrlPath (String url, int timeoutMs , int limit ) throws IOException
+ {
+ final StringBuffer contexturl= new StringBuffer(url);
+ // final StringBuffer url = new StringBuffer ( CambriaConstants.makeConsumerUrl ( host, fTopic, fGroup, fId ) );
+ final StringBuffer adds = new StringBuffer ();
+ if ( timeoutMs > -1 ) adds.append ( "timeout=" ).append ( timeoutMs );
+ if ( limit > -1 )
+ {
+ if ( adds.length () > 0 )
+ {
+ adds.append ( "&" );
+ }
+ adds.append ( "limit=" ).append ( limit );
+ }
+ if ( fFilter != null && fFilter.length () > 0 )
+ {
+ try {
+ if ( adds.length () > 0 )
+ {
+ adds.append ( "&" );
+ }
+ adds.append("filter=").append(URLEncoder.encode(fFilter, "UTF-8"));
+ } catch (UnsupportedEncodingException e) {
+ throw new RuntimeException(e.getMessage() + "....say whaaaat?!");
+ }
+ }
+ if ( adds.length () > 0 )
+ {
+ contexturl.append ( "?" ).append ( adds.toString () );
+ }
+
+ //sender.setSubContext(url.toString());
+ return contexturl.toString ();
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public void setUsername(String username) {
+ this.username = username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public void setHost(String host) {
+ this.host = host;
+ }
+
+ public String getAuthKey() {
+ return authKey;
+ }
+
+ public void setAuthKey(String authKey) {
+ this.authKey = authKey;
+ }
+
+ public String getAuthDate() {
+ return authDate;
+ }
+
+ public void setAuthDate(String authDate) {
+ this.authDate = authDate;
+ }
+
+ public String getfFilter() {
+ return fFilter;
+ }
+
+ public void setfFilter(String fFilter) {
+ this.fFilter = fFilter;
+ }
+
+ private String readRoute(String routeKey) {
+
+ try {
+
+ MRClientFactory.prop.load(new FileReader(new File (MRClientFactory.routeFilePath)));
+
+ } catch (Exception ex) {
+ fLog.error("Reply Router Error " + ex.toString() );
+ }
+ String routeOffer = MRClientFactory.prop.getProperty(routeKey);
+ return routeOffer;
+ }
+
+ @Override
+ public MRConsumerResponse fetchWithReturnConsumerResponse() {
+
+ // fetch with the timeout and limit set in constructor
+ return fetchWithReturnConsumerResponse(fTimeoutMs, fLimit);
+ }
+
+ @Override
+ public MRConsumerResponse fetchWithReturnConsumerResponse(int timeoutMs,
+ int limit) {
+ final LinkedList<String> msgs = new LinkedList<String>();
+ MRConsumerResponse mrConsumerResponse = new MRConsumerResponse();
+ try {
+ if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(
+ protocolFlag)) {
+ DMEConfigure(timeoutMs, limit);
+
+ String reply = sender.sendAndWait(timeoutMs + 10000L);
+
+ final JSONObject o = getResponseDataInJsonWithResponseReturned(reply);
+
+ if (o != null) {
+ final JSONArray a = o.getJSONArray("result");
+
+ if (a != null) {
+ for (int i = 0; i < a.length(); i++) {
+ if (a.get(i) instanceof String)
+ msgs.add(a.getString(i));
+ else
+ msgs.add(a.getJSONObject(i).toString());
+
+ }
+ }
+
+ }
+ createMRConsumerResponse(reply, mrConsumerResponse);
+ }
+
+ if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(
+ protocolFlag)) {
+ final String urlPath = createUrlPath(
+ MRConstants.makeConsumerUrl(host, fTopic, fGroup, fId,
+ props.getProperty("Protocol")), timeoutMs,
+ limit);
+
+ String response = getResponse(urlPath, username, password,
+ protocolFlag);
+
+ final JSONObject o = getResponseDataInJsonWithResponseReturned(response);
+
+ if (o != null) {
+ final JSONArray a = o.getJSONArray("result");
+
+ if (a != null) {
+ for (int i = 0; i < a.length(); i++) {
+ if (a.get(i) instanceof String)
+ msgs.add(a.getString(i));
+ else
+ msgs.add(a.getJSONObject(i).toString());
+
+ }
+ }
+
+ }
+ createMRConsumerResponse(response, mrConsumerResponse);
+ }
+
+ if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(
+ protocolFlag)) {
+ final String urlPath = createUrlPath(
+ MRConstants.makeConsumerUrl(host, fTopic, fGroup, fId,
+ props.getProperty("Protocol")), timeoutMs,
+ limit);
+
+ String response = getAuthResponse(urlPath, authKey, authDate,
+ username, password, protocolFlag);
+ final JSONObject o = getResponseDataInJsonWithResponseReturned(response);
+ if (o != null) {
+ final JSONArray a = o.getJSONArray("result");
+
+ if (a != null) {
+ for (int i = 0; i < a.length(); i++) {
+ if (a.get(i) instanceof String)
+ msgs.add(a.getString(i));
+ else
+ msgs.add(a.getJSONObject(i).toString());
+
+ }
+ }
+
+ }
+ createMRConsumerResponse(response, mrConsumerResponse);
+ }
+
+
+
+ } catch (JSONException e) {
+ mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+ mrConsumerResponse.setResponseMessage(e.getMessage());
+ } catch (HttpException e) {
+ mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+ mrConsumerResponse.setResponseMessage(e.getMessage());
+ }catch(DME2Exception e){
+ mrConsumerResponse.setResponseCode(e.getErrorCode());
+ mrConsumerResponse.setResponseMessage(e.getErrorMessage());
+ }catch (Exception e) {
+ mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+ mrConsumerResponse.setResponseMessage(e.getMessage());
+ }
+ mrConsumerResponse.setActualMessages(msgs);
+ return mrConsumerResponse;
+ }
+
+ private void createMRConsumerResponse(String reply, MRConsumerResponse mrConsumerResponse) {
+
+ if(reply.startsWith("{")){
+ JSONObject jObject = new JSONObject(reply);
+ String message = jObject.getString("message");
+ int status = jObject.getInt("status");
+
+ mrConsumerResponse.setResponseCode(Integer.toString(status));
+
+ if(null != message){
+ mrConsumerResponse.setResponseMessage(message);
+ }
+ }else if (reply.startsWith("<")){
+ mrConsumerResponse.setResponseCode(getHTTPErrorResponseCode(reply));
+ mrConsumerResponse.setResponseMessage(getHTTPErrorResponseMessage(reply));
+ }else{
+ mrConsumerResponse.setResponseCode(String.valueOf(HttpStatus.SC_OK));
+ mrConsumerResponse.setResponseMessage(SUCCESS_MESSAGE);
+ }
+
+ }
+
+
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRFormat.java b/src/main/java/com/att/nsa/mr/client/impl/MRFormat.java
new file mode 100644
index 0000000..09a317b
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRFormat.java
@@ -0,0 +1,49 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client.impl;
+
+enum MRFormat
+{
+ /**
+ * Messages are sent using MR's message format.
+ */
+ CAMBRIA
+ {
+ public String toString() { return "application/cambria"; }
+ },
+
+ /**
+ * Messages are sent using MR's message format with compression.
+ */
+ CAMBRIA_ZIP
+ {
+ public String toString() { return "application/cambria-zip"; }
+ },
+
+ /**
+ * messages are sent as simple JSON objects.
+ */
+ JSON
+ {
+ public String toString() { return "application/json"; }
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRMetaClient.java b/src/main/java/com/att/nsa/mr/client/impl/MRMetaClient.java
new file mode 100644
index 0000000..d32a7ef
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRMetaClient.java
@@ -0,0 +1,260 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client.impl;
+
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.util.Collection;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+
+import com.att.nsa.apiClient.credentials.ApiCredential;
+import com.att.nsa.apiClient.http.HttpException;
+import com.att.nsa.apiClient.http.HttpObjectNotFoundException;
+import com.att.nsa.mr.client.MRIdentityManager;
+import com.att.nsa.mr.client.MRTopicManager;
+
+public class MRMetaClient extends MRBaseClient implements MRTopicManager, MRIdentityManager
+{
+ public MRMetaClient ( Collection<String> baseUrls ) throws MalformedURLException
+ {
+ super ( baseUrls );
+ }
+
+ @Override
+ public Set<String> getTopics () throws IOException
+ {
+ final TreeSet<String> set = new TreeSet<String> ();
+ try
+ {
+ final JSONObject topicSet = get ( "/topics" );
+ final JSONArray a = topicSet.getJSONArray ( "topics" );
+ for ( int i=0; i<a.length (); i++ )
+ {
+ set.add ( a.getString ( i ) );
+ }
+ }
+ catch ( HttpObjectNotFoundException e )
+ {
+ getLog().warn ( "No /topics endpoint on service." );
+ }
+ catch ( JSONException e )
+ {
+ getLog().warn ( "Bad /topics result from service." );
+ }
+ catch ( HttpException e )
+ {
+ throw new IOException ( e );
+ }
+ return set;
+ }
+
+ @Override
+ public TopicInfo getTopicMetadata ( String topic ) throws HttpObjectNotFoundException, IOException
+ {
+ try
+ {
+ final JSONObject topicData = get ( "/topics/" + MRConstants.escape ( topic ) );
+ return new TopicInfo ()
+ {
+ @Override
+ public String getOwner ()
+ {
+ return topicData.optString ( "owner", null );
+ }
+
+ @Override
+ public String getDescription ()
+ {
+ return topicData.optString ( "description", null );
+ }
+
+ @Override
+ public Set<String> getAllowedProducers ()
+ {
+ final JSONObject acl = topicData.optJSONObject ( "writerAcl" );
+ if ( acl != null && acl.optBoolean ( "enabled", true ) )
+ {
+ return jsonArrayToSet ( acl.optJSONArray ( "users" ) );
+ }
+ return null;
+ }
+
+ @Override
+ public Set<String> getAllowedConsumers ()
+ {
+ final JSONObject acl = topicData.optJSONObject ( "readerAcl" );
+ if ( acl != null && acl.optBoolean ( "enabled", true ) )
+ {
+ return jsonArrayToSet ( acl.optJSONArray ( "users" ) );
+ }
+ return null;
+ }
+ };
+ }
+ catch ( JSONException e )
+ {
+ throw new IOException ( e );
+ }
+ catch ( HttpException e )
+ {
+ throw new IOException ( e );
+ }
+ }
+
+ @Override
+ public void createTopic ( String topicName, String topicDescription, int partitionCount, int replicationCount ) throws HttpException, IOException
+ {
+ final JSONObject o = new JSONObject ();
+ o.put ( "topicName", topicName );
+ o.put ( "topicDescription", topicDescription );
+ o.put ( "partitionCount", partitionCount );
+ o.put ( "replicationCount", replicationCount );
+ post ( "/topics/create", o, false );
+ }
+
+ @Override
+ public void deleteTopic ( String topic ) throws HttpException, IOException
+ {
+ delete ( "/topics/" + MRConstants.escape ( topic ) );
+ }
+
+ @Override
+ public boolean isOpenForProducing ( String topic ) throws HttpObjectNotFoundException, IOException
+ {
+ return null == getAllowedProducers ( topic );
+ }
+
+ @Override
+ public Set<String> getAllowedProducers ( String topic ) throws HttpObjectNotFoundException, IOException
+ {
+ return getTopicMetadata ( topic ).getAllowedProducers ();
+ }
+
+ @Override
+ public void allowProducer ( String topic, String apiKey ) throws HttpObjectNotFoundException, HttpException, IOException
+ {
+ put ( "/topics/" + MRConstants.escape ( topic ) + "/producers/" + MRConstants.escape ( apiKey ), new JSONObject() );
+ }
+
+ @Override
+ public void revokeProducer ( String topic, String apiKey ) throws HttpException, IOException
+ {
+ delete ( "/topics/" + MRConstants.escape ( topic ) + "/producers/" + MRConstants.escape ( apiKey ) );
+ }
+
+ @Override
+ public boolean isOpenForConsuming ( String topic ) throws HttpObjectNotFoundException, IOException
+ {
+ return null == getAllowedConsumers ( topic );
+ }
+
+ @Override
+ public Set<String> getAllowedConsumers ( String topic ) throws HttpObjectNotFoundException, IOException
+ {
+ return getTopicMetadata ( topic ).getAllowedConsumers ();
+ }
+
+ @Override
+ public void allowConsumer ( String topic, String apiKey ) throws HttpObjectNotFoundException, HttpException, IOException
+ {
+ put ( "/topics/" + MRConstants.escape ( topic ) + "/consumers/" + MRConstants.escape ( apiKey ), new JSONObject() );
+ }
+
+ @Override
+ public void revokeConsumer ( String topic, String apiKey ) throws HttpException, IOException
+ {
+ delete ( "/topics/" + MRConstants.escape ( topic ) + "/consumers/" + MRConstants.escape ( apiKey ) );
+ }
+
+ @Override
+ public ApiCredential createApiKey ( String email, String description ) throws HttpException, MRApiException, IOException
+ {
+ try
+ {
+ final JSONObject o = new JSONObject ();
+ o.put ( "email", email );
+ o.put ( "description", description );
+ final JSONObject reply = post ( "/apiKeys/create", o, true );
+ return new ApiCredential ( reply.getString ( "key" ), reply.getString ( "secret" ) );
+ }
+ catch ( JSONException e )
+ {
+ // the response doesn't meet our expectation
+ throw new MRApiException ( "The API key response is incomplete.", e );
+ }
+ }
+
+ @Override
+ public ApiKey getApiKey ( String apiKey ) throws HttpObjectNotFoundException, HttpException, IOException
+ {
+ final JSONObject keyEntry = get ( "/apiKeys/" + MRConstants.escape ( apiKey ) );
+ if ( keyEntry == null )
+ {
+ return null;
+ }
+
+ return new ApiKey ()
+ {
+ @Override
+ public String getEmail ()
+ {
+ final JSONObject aux = keyEntry.optJSONObject ( "aux" );
+ if ( aux != null )
+ {
+ return aux.optString ( "email" );
+ }
+ return null;
+ }
+
+ @Override
+ public String getDescription ()
+ {
+ final JSONObject aux = keyEntry.optJSONObject ( "aux" );
+ if ( aux != null )
+ {
+ return aux.optString ( "description" );
+ }
+ return null;
+ }
+ };
+ }
+
+ @Override
+ public void updateCurrentApiKey ( String email, String description ) throws HttpObjectNotFoundException, HttpException, IOException
+ {
+ final JSONObject o = new JSONObject ();
+ if ( email != null ) o.put ( "email", email );
+ if ( description != null ) o.put ( "description", description );
+ patch ( "/apiKeys/" + MRConstants.escape ( getCurrentApiKey() ), o );
+ }
+
+ @Override
+ public void deleteCurrentApiKey () throws HttpException, IOException
+ {
+ delete ( "/apiKeys/" + MRConstants.escape ( getCurrentApiKey() ) );
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java b/src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java
new file mode 100644
index 0000000..ad925b1
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java
@@ -0,0 +1,939 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client.impl;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.zip.GZIPOutputStream;
+
+import javax.ws.rs.core.MultivaluedMap;
+
+import org.apache.http.HttpException;
+import org.apache.http.HttpStatus;
+import org.json.JSONArray;
+import org.json.JSONObject;
+
+import com.att.aft.dme2.api.DME2Client;
+import com.att.aft.dme2.api.DME2Exception;
+import com.att.nsa.mr.client.HostSelector;
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.response.MRPublisherResponse;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+public class MRSimplerBatchPublisher extends MRBaseClient implements MRBatchingPublisher
+{
+ public static class Builder
+ {
+ public Builder ()
+ {
+ }
+
+ public Builder againstUrls ( Collection<String> baseUrls )
+ {
+ fUrls = baseUrls;
+ return this;
+ }
+
+ public Builder onTopic ( String topic )
+ {
+ fTopic = topic;
+ return this;
+ }
+
+ public Builder batchTo ( int maxBatchSize, long maxBatchAgeMs )
+ {
+ fMaxBatchSize = maxBatchSize;
+ fMaxBatchAgeMs = maxBatchAgeMs;
+ return this;
+ }
+
+ public Builder compress ( boolean compress )
+ {
+ fCompress = compress;
+ return this;
+ }
+
+ public Builder httpThreadTime ( int threadOccuranceTime )
+ {
+ this.threadOccuranceTime = threadOccuranceTime;
+ return this;
+ }
+
+ public Builder allowSelfSignedCertificates( boolean allowSelfSignedCerts )
+ {
+ fAllowSelfSignedCerts = allowSelfSignedCerts;
+ return this;
+ }
+
+ public Builder withResponse ( boolean withResponse)
+ {
+ fWithResponse = withResponse;
+ return this;
+ }
+ public MRSimplerBatchPublisher build ()
+ {
+ if(!fWithResponse)
+ {
+ try {
+ return new MRSimplerBatchPublisher ( fUrls, fTopic, fMaxBatchSize, fMaxBatchAgeMs, fCompress, fAllowSelfSignedCerts,threadOccuranceTime);
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ } else
+ {
+ try {
+ return new MRSimplerBatchPublisher ( fUrls, fTopic, fMaxBatchSize, fMaxBatchAgeMs, fCompress, fAllowSelfSignedCerts, fMaxBatchSize);
+ } catch (MalformedURLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ }
+
+ private Collection<String> fUrls;
+ private String fTopic;
+ private int fMaxBatchSize = 100;
+ private long fMaxBatchAgeMs = 1000;
+ private boolean fCompress = false;
+ private int threadOccuranceTime = 50;
+ private boolean fAllowSelfSignedCerts = false;
+ private boolean fWithResponse = false;
+
+ };
+
+ @Override
+ public int send ( String partition, String msg )
+ {
+ return send ( new message ( partition, msg ) );
+ }
+ @Override
+ public int send ( String msg )
+ {
+ return send ( new message ( null, msg ) );
+ }
+
+
+ @Override
+ public int send ( message msg )
+ {
+ final LinkedList<message> list = new LinkedList<message> ();
+ list.add ( msg );
+ return send ( list );
+ }
+
+
+
+ @Override
+ public synchronized int send ( Collection<message> msgs )
+ {
+ if ( fClosed )
+ {
+ throw new IllegalStateException ( "The publisher was closed." );
+ }
+
+ for ( message userMsg : msgs )
+ {
+ fPending.add ( new TimestampedMessage ( userMsg ) );
+ }
+ return getPendingMessageCount ();
+ }
+
+ @Override
+ public synchronized int getPendingMessageCount ()
+ {
+ return fPending.size ();
+ }
+
+ @Override
+ public void close ()
+ {
+ try
+ {
+ final List<message> remains = close ( Long.MAX_VALUE, TimeUnit.MILLISECONDS );
+ if ( remains.size() > 0 )
+ {
+ getLog().warn ( "Closing publisher with " + remains.size() + " messages unsent. "
+ + "Consider using MRBatchingPublisher.close( long timeout, TimeUnit timeoutUnits ) to recapture unsent messages on close." );
+ }
+ }
+ catch ( InterruptedException e )
+ {
+ getLog().warn ( "Possible message loss. " + e.getMessage(), e );
+ }
+ catch ( IOException e )
+ {
+ getLog().warn ( "Possible message loss. " + e.getMessage(), e );
+ }
+ }
+
+ @Override
+ public List<message> close ( long time, TimeUnit unit ) throws IOException, InterruptedException
+ {
+ synchronized ( this )
+ {
+ fClosed = true;
+
+ // stop the background sender
+ fExec.setContinueExistingPeriodicTasksAfterShutdownPolicy ( false );
+ fExec.setExecuteExistingDelayedTasksAfterShutdownPolicy ( false );
+ fExec.shutdown ();
+ }
+
+ final long now = Clock.now ();
+ final long waitInMs = TimeUnit.MILLISECONDS.convert ( time, unit );
+ final long timeoutAtMs = now + waitInMs;
+
+ while ( Clock.now() < timeoutAtMs && getPendingMessageCount() > 0 )
+ {
+ send ( true );
+ Thread.sleep ( 250 );
+ }
+
+ synchronized ( this )
+ {
+ final LinkedList<message> result = new LinkedList<message> ();
+ fPending.drainTo ( result );
+ return result;
+ }
+ }
+
+ /**
+ * Possibly send a batch to the MR server. This is called by the background thread
+ * and the close() method
+ *
+ * @param force
+ */
+ private synchronized void send ( boolean force )
+ {
+ if ( force || shouldSendNow () )
+ {
+ if ( !sendBatch () )
+ {
+ getLog().warn ( "Send failed, " + fPending.size() + " message to send." );
+
+ // note the time for back-off
+ fDontSendUntilMs = sfWaitAfterError + Clock.now ();
+ }
+ }
+ }
+
+ private synchronized boolean shouldSendNow ()
+ {
+ boolean shouldSend = false;
+ if ( fPending.size () > 0 )
+ {
+ final long nowMs = Clock.now ();
+
+ shouldSend = ( fPending.size() >= fMaxBatchSize );
+ if ( !shouldSend )
+ {
+ final long sendAtMs = fPending.peek().timestamp + fMaxBatchAgeMs;
+ shouldSend = sendAtMs <= nowMs;
+ }
+
+ // however, wait after an error
+ shouldSend = shouldSend && nowMs >= fDontSendUntilMs;
+ }
+ return shouldSend;
+ }
+
+ private synchronized boolean sendBatch ()
+ {
+ // it's possible for this call to be made with an empty list. in this case, just return.
+ if ( fPending.size() < 1 )
+ {
+ return true;
+ }
+
+ final long nowMs = Clock.now ();
+
+ host = this.fHostSelector.selectBaseHost();
+
+ final String httpurl = MRConstants.makeUrl ( host, fTopic,props.getProperty("Protocol"),props.getProperty("partition") );
+
+
+ try
+ {
+ /*final String contentType =
+ fCompress ?
+ MRFormat.CAMBRIA_ZIP.toString () :
+ MRFormat.CAMBRIA.toString ()
+ ;*/
+
+ final ByteArrayOutputStream baseStream = new ByteArrayOutputStream ();
+ OutputStream os = baseStream;
+ final String contentType = props.getProperty("contenttype");
+ if(contentType.equalsIgnoreCase("application/json")){
+ JSONArray jsonArray = new JSONArray();
+ for ( TimestampedMessage m : fPending )
+ {
+ JSONObject jsonObject = new JSONObject(m.fMsg);
+
+ jsonArray.put(jsonObject);
+
+ }
+ os.write (jsonArray.toString().getBytes() );
+ os.close();
+
+ }else if (contentType.equalsIgnoreCase("text/plain")){
+ for ( TimestampedMessage m : fPending )
+ {
+ os.write ( m.fMsg.getBytes() );
+ os.write ( '\n' );
+ }
+ os.close ();
+ } else if (contentType.equalsIgnoreCase("application/cambria") || (contentType.equalsIgnoreCase("application/cambria-zip"))){
+ if ( contentType.equalsIgnoreCase("application/cambria-zip") )
+ {
+ os = new GZIPOutputStream ( baseStream );
+ }
+ for ( TimestampedMessage m : fPending )
+ {
+
+ os.write ( ( "" + m.fPartition.length () ).getBytes() );
+ os.write ( '.' );
+ os.write ( ( "" + m.fMsg.length () ).getBytes() );
+ os.write ( '.' );
+ os.write ( m.fPartition.getBytes() );
+ os.write ( m.fMsg.getBytes() );
+ os.write ( '\n' );
+ }
+ os.close ();
+ }else{
+ for ( TimestampedMessage m : fPending )
+ {
+ os.write ( m.fMsg.getBytes() );
+
+ }
+ os.close ();
+ }
+
+
+
+ final long startMs = Clock.now ();
+ if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) {
+
+
+ DME2Configue();
+
+ Thread.sleep(5);
+ getLog().info ( "sending " + fPending.size() + " msgs to " + url+subContextPath + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms" );
+ sender.setPayload(os.toString());
+ String dmeResponse = sender.sendAndWait(5000L);
+
+ final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):"
+ + dmeResponse.toString();
+ getLog().info(logLine);
+ fPending.clear();
+ return true;
+ }
+
+ if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
+ getLog().info ( "sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms" );
+ final JSONObject result = postAuth(httpurl, baseStream.toByteArray(), contentType, authKey, authDate, username, password,protocolFlag);
+ //System.out.println(result.getInt("status"));
+ //Here we are checking for error response. If HTTP status
+ //code is not within the http success response code
+ //then we consider this as error and return false
+ if(result.getInt("status") < 200 || result.getInt("status") > 299) {
+ return false;
+ }
+ final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):" + result.toString();
+ getLog().info(logLine);
+ fPending.clear();
+ return true;
+ }
+
+ if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) {
+ getLog().info ( "sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms" );
+ final JSONObject result = post(httpurl, baseStream.toByteArray(), contentType, username, password, protocolFlag);
+
+
+ //System.out.println(result.getInt("status"));
+ //Here we are checking for error response. If HTTP status
+ //code is not within the http success response code
+ //then we consider this as error and return false
+ if(result.getInt("status") < 200 || result.getInt("status") > 299) {
+ return false;
+ }
+ final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):" + result.toString();
+ getLog().info(logLine);
+ fPending.clear();
+ return true;
+ }
+ }
+ catch ( IllegalArgumentException x ) {
+ getLog().warn ( x.getMessage(), x );
+ } catch ( IOException x ) {
+ getLog().warn ( x.getMessage(), x );
+ } catch (HttpException x) {
+ getLog().warn ( x.getMessage(), x );
+ } catch (Exception x) {
+ getLog().warn(x.getMessage(), x);
+ }
+ return false;
+ }
+
+ public synchronized MRPublisherResponse sendBatchWithResponse ()
+ {
+ // it's possible for this call to be made with an empty list. in this case, just return.
+ if ( fPending.size() < 1 )
+ {
+ pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
+ pubResponse.setResponseMessage("No Messages to send");
+ return pubResponse;
+ }
+
+ final long nowMs = Clock.now ();
+
+ host = this.fHostSelector.selectBaseHost();
+
+ final String httpurl = MRConstants.makeUrl ( host, fTopic,props.getProperty("Protocol"),props.getProperty("partition") );
+ OutputStream os=null;
+ try
+ {
+
+ final ByteArrayOutputStream baseStream = new ByteArrayOutputStream ();
+ os = baseStream;
+ final String contentType = props.getProperty("contenttype");
+ if(contentType.equalsIgnoreCase("application/json")){
+ JSONArray jsonArray = new JSONArray();
+ for ( TimestampedMessage m : fPending )
+ {
+ JSONObject jsonObject = new JSONObject(m.fMsg);
+
+ jsonArray.put(jsonObject);
+
+ }
+ os.write (jsonArray.toString().getBytes() );
+ }else if (contentType.equalsIgnoreCase("text/plain")){
+ for ( TimestampedMessage m : fPending )
+ {
+ os.write ( m.fMsg.getBytes() );
+ os.write ( '\n' );
+ }
+ } else if (contentType.equalsIgnoreCase("application/cambria") || (contentType.equalsIgnoreCase("application/cambria-zip"))){
+ if ( contentType.equalsIgnoreCase("application/cambria-zip") )
+ {
+ os = new GZIPOutputStream ( baseStream );
+ }
+ for ( TimestampedMessage m : fPending )
+ {
+
+ os.write ( ( "" + m.fPartition.length () ).getBytes() );
+ os.write ( '.' );
+ os.write ( ( "" + m.fMsg.length () ).getBytes() );
+ os.write ( '.' );
+ os.write ( m.fPartition.getBytes() );
+ os.write ( m.fMsg.getBytes() );
+ os.write ( '\n' );
+ }
+ os.close ();
+ }else{
+ for ( TimestampedMessage m : fPending )
+ {
+ os.write ( m.fMsg.getBytes() );
+
+ }
+ }
+
+
+
+ final long startMs = Clock.now ();
+ if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) {
+
+
+ try {
+ DME2Configue();
+
+ Thread.sleep(5);
+ getLog().info ( "sending " + fPending.size() + " msgs to " + url+subContextPath + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms" );
+ sender.setPayload(os.toString());
+
+
+ String dmeResponse = sender.sendAndWait(5000L);
+ System.out.println("dmeres->"+dmeResponse);
+
+
+ pubResponse = createMRPublisherResponse(dmeResponse,pubResponse);
+
+ if(Integer.valueOf(pubResponse.getResponseCode()) < 200 || Integer.valueOf(pubResponse.getResponseCode()) > 299) {
+
+ return pubResponse;
+ }
+ final String logLine = String.valueOf((Clock.now() - startMs))
+ + dmeResponse.toString();
+ getLog().info(logLine);
+ fPending.clear();
+
+ }
+ catch (DME2Exception x) {
+ getLog().warn(x.getMessage(), x);
+ pubResponse.setResponseCode(x.getErrorCode());
+ pubResponse.setResponseMessage(x.getErrorMessage());
+ } catch (URISyntaxException x) {
+
+ getLog().warn(x.getMessage(), x);
+ pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
+ pubResponse.setResponseMessage(x.getMessage());
+ } catch (Exception x) {
+
+ pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+ pubResponse.setResponseMessage(x.getMessage());
+
+ }
+
+ return pubResponse;
+ }
+
+ if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
+ getLog().info ( "sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms" );
+ final String result = postAuthwithResponse(httpurl, baseStream.toByteArray(), contentType, authKey, authDate, username, password,protocolFlag);
+ //System.out.println(result.getInt("status"));
+ //Here we are checking for error response. If HTTP status
+ //code is not within the http success response code
+ //then we consider this as error and return false
+
+
+ pubResponse = createMRPublisherResponse(result,pubResponse);
+
+ if(Integer.valueOf(pubResponse.getResponseCode()) < 200 || Integer.valueOf(pubResponse.getResponseCode()) > 299) {
+
+ return pubResponse;
+ }
+
+ final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):" + result.toString();
+ getLog().info(logLine);
+ fPending.clear();
+ return pubResponse;
+ }
+
+ if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) {
+ getLog().info ( "sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms" );
+ final String result = postWithResponse(httpurl, baseStream.toByteArray(), contentType, username, password, protocolFlag);
+
+ //System.out.println(result.getInt("status"));
+ //Here we are checking for error response. If HTTP status
+ //code is not within the http success response code
+ //then we consider this as error and return false
+ pubResponse = createMRPublisherResponse(result,pubResponse);
+
+ if(Integer.valueOf(pubResponse.getResponseCode()) < 200 || Integer.valueOf(pubResponse.getResponseCode()) > 299) {
+
+ return pubResponse;
+ }
+
+ final String logLine = String.valueOf((Clock.now() - startMs));
+ getLog().info(logLine);
+ fPending.clear();
+ return pubResponse;
+ }
+ }
+ catch ( IllegalArgumentException x ) {
+ getLog().warn ( x.getMessage(), x );
+ pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
+ pubResponse.setResponseMessage(x.getMessage());
+
+ } catch ( IOException x ) {
+ getLog().warn ( x.getMessage(), x );
+ pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+ pubResponse.setResponseMessage(x.getMessage());
+
+ } catch (HttpException x) {
+ getLog().warn ( x.getMessage(), x );
+ pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
+ pubResponse.setResponseMessage(x.getMessage());
+
+ } catch (Exception x) {
+ getLog().warn(x.getMessage(), x);
+
+ pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+ pubResponse.setResponseMessage(x.getMessage());
+
+ }
+
+ finally {
+ if (fPending.size()>0) {
+ getLog().warn ( "Send failed, " + fPending.size() + " message to send." );
+ pubResponse.setPendingMsgs(fPending.size());
+ }
+ if (os != null) {
+ try {
+ os.close();
+ } catch (Exception x) {
+ getLog().warn(x.getMessage(), x);
+ pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+ pubResponse.setResponseMessage("Error in closing Output Stream");
+ }
+ }
+ }
+
+ return pubResponse;
+ }
+
+private MRPublisherResponse createMRPublisherResponse(String reply, MRPublisherResponse mrPubResponse) {
+
+ if (reply.isEmpty())
+ {
+
+ mrPubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
+ mrPubResponse.setResponseMessage("Please verify the Producer properties");
+ }
+ else if(reply.startsWith("{"))
+ {
+ JSONObject jObject = new JSONObject(reply);
+ if(jObject.has("message") && jObject.has("status"))
+ {
+ String message = jObject.getString("message");
+ if(null != message)
+ {
+ mrPubResponse.setResponseMessage(message);
+ }
+ mrPubResponse.setResponseCode(Integer.toString(jObject.getInt("status")));
+ }
+ else
+ {
+ mrPubResponse.setResponseCode(String.valueOf(HttpStatus.SC_OK));
+ mrPubResponse.setResponseMessage(reply);
+ }
+ }
+ else if (reply.startsWith("<"))
+ {
+ String responseCode = getHTTPErrorResponseCode(reply);
+ if( responseCode.contains("403"))
+ {
+ responseCode = "403";
+ }
+ mrPubResponse.setResponseCode(responseCode);
+ mrPubResponse.setResponseMessage(getHTTPErrorResponseMessage(reply));
+ }
+
+ return mrPubResponse;
+ }
+
+ private final String fTopic;
+ private final int fMaxBatchSize;
+ private final long fMaxBatchAgeMs;
+ private final boolean fCompress;
+ private int threadOccuranceTime;
+ private boolean fClosed;
+ private String username;
+ private String password;
+ private String host;
+
+ //host selector
+ private HostSelector fHostSelector = null;
+
+ private final LinkedBlockingQueue<TimestampedMessage> fPending;
+ private long fDontSendUntilMs;
+ private final ScheduledThreadPoolExecutor fExec;
+
+ private String latitude;
+ private String longitude;
+ private String version;
+ private String serviceName;
+ private String env;
+ private String partner;
+ private String routeOffer;
+ private String subContextPath;
+ private String protocol;
+ private String methodType;
+ private String url;
+ private String dmeuser;
+ private String dmepassword;
+ private String contentType;
+ private static final long sfWaitAfterError = 10000;
+ private HashMap<String, String> DMETimeOuts;
+ private DME2Client sender;
+ public String protocolFlag = ProtocolTypeConstants.DME2.getValue();
+ public String producerFilePath;
+ private String authKey;
+ private String authDate;
+ private String handlers;
+ private Properties props;
+ public static String routerFilePath;
+ public static Map<String, String> headers=new HashMap<String, String>();
+ public static MultivaluedMap<String, Object> headersMap;
+
+
+ private MRPublisherResponse pubResponse;
+
+ public MRPublisherResponse getPubResponse() {
+ return pubResponse;
+ }
+ public void setPubResponse(MRPublisherResponse pubResponse) {
+ this.pubResponse = pubResponse;
+ }
+
+ public static String getRouterFilePath() {
+ return routerFilePath;
+ }
+
+ public static void setRouterFilePath(String routerFilePath) {
+ MRSimplerBatchPublisher.routerFilePath = routerFilePath;
+ }
+
+ public Properties getProps() {
+ return props;
+ }
+
+ public void setProps(Properties props) {
+ this.props = props;
+ }
+
+ public String getProducerFilePath() {
+ return producerFilePath;
+ }
+
+ public void setProducerFilePath(String producerFilePath) {
+ this.producerFilePath = producerFilePath;
+ }
+
+ public String getProtocolFlag() {
+ return protocolFlag;
+ }
+
+ public void setProtocolFlag(String protocolFlag) {
+ this.protocolFlag = protocolFlag;
+ }
+
+
+ private void DME2Configue() throws Exception {
+ try {
+
+ /* FileReader reader = new FileReader(new File (producerFilePath));
+ Properties props = new Properties();
+ props.load(reader);*/
+ latitude = props.getProperty("Latitude");
+ longitude = props.getProperty("Longitude");
+ version = props.getProperty("Version");
+ serviceName = props.getProperty("ServiceName");
+ env = props.getProperty("Environment");
+ partner = props.getProperty("Partner");
+ routeOffer = props.getProperty("routeOffer");
+ subContextPath = props.getProperty("SubContextPath")+fTopic;
+ /*if(props.getProperty("partition")!=null && !props.getProperty("partition").equalsIgnoreCase("")){
+ subContextPath=subContextPath+"?partitionKey="+props.getProperty("partition");
+ }*/
+ protocol = props.getProperty("Protocol");
+ methodType = props.getProperty("MethodType");
+ dmeuser = props.getProperty("username");
+ dmepassword = props.getProperty("password");
+ contentType = props.getProperty("contenttype");
+ handlers = props.getProperty("sessionstickinessrequired");
+ routerFilePath= props.getProperty("DME2preferredRouterFilePath");
+
+ /**
+ * Changes to DME2Client url to use Partner for auto failover between data centers
+ * When Partner value is not provided use the routeOffer value for auto failover within a cluster
+ */
+
+
+ String partitionKey = props.getProperty("partition");
+
+ if (partner != null && !partner.isEmpty() )
+ {
+ url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&partner="+partner;
+ if(partitionKey!=null && !partitionKey.equalsIgnoreCase("")){
+ url = url + "&partitionKey=" + partitionKey;
+ }
+ }
+ else if (routeOffer!=null && !routeOffer.isEmpty())
+ {
+ url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&routeoffer="+routeOffer;
+ if(partitionKey!=null && !partitionKey.equalsIgnoreCase("")){
+ url = url + "&partitionKey=" + partitionKey;
+ }
+ }
+
+ DMETimeOuts = new HashMap<String, String>();
+ DMETimeOuts.put("AFT_DME2_EP_READ_TIMEOUT_MS", props.getProperty("AFT_DME2_EP_READ_TIMEOUT_MS"));
+ DMETimeOuts.put("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", props.getProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS"));
+ DMETimeOuts.put("AFT_DME2_EP_CONN_TIMEOUT", props.getProperty("AFT_DME2_EP_CONN_TIMEOUT"));
+ DMETimeOuts.put("Content-Type", contentType);
+ System.setProperty("AFT_LATITUDE", latitude);
+ System.setProperty("AFT_LONGITUDE", longitude);
+ System.setProperty("AFT_ENVIRONMENT",props.getProperty("AFT_ENVIRONMENT"));
+ //System.setProperty("DME2.DEBUG", "true");
+ // System.setProperty("AFT_DME2_HTTP_EXCHANGE_TRACE_ON", "true");
+ //System.out.println("XXXXXX"+url);
+
+ //SSL changes
+ System.setProperty("AFT_DME2_CLIENT_SSL_INCLUDE_PROTOCOLS",
+ "SSLv3,TLSv1,TLSv1.1");
+ System.setProperty("AFT_DME2_CLIENT_IGNORE_SSL_CONFIG", "false");
+ System.setProperty("AFT_DME2_CLIENT_KEYSTORE_PASSWORD", "changeit");
+
+ //SSL changes
+
+ sender = new DME2Client(new URI(url), 5000L);
+
+ sender.setAllowAllHttpReturnCodes(true);
+ sender.setMethod(methodType);
+ sender.setSubContext(subContextPath);
+ sender.setCredentials(dmeuser, dmepassword);
+ sender.setHeaders(DMETimeOuts);
+ if(handlers.equalsIgnoreCase("yes")){
+ sender.addHeader("AFT_DME2_EXCHANGE_REQUEST_HANDLERS", props.getProperty("AFT_DME2_EXCHANGE_REQUEST_HANDLERS"));
+ sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", props.getProperty("AFT_DME2_EXCHANGE_REPLY_HANDLERS"));
+ sender.addHeader("AFT_DME2_REQ_TRACE_ON", props.getProperty("AFT_DME2_REQ_TRACE_ON"));
+ }else{
+ sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", "com.att.nsa.mr.dme.client.HeaderReplyHandler");
+ }
+ } catch (DME2Exception x) {
+ getLog().warn(x.getMessage(), x);
+ throw new DME2Exception(x.getErrorCode(),x.getErrorMessage());
+ } catch (URISyntaxException x) {
+
+ getLog().warn(x.getMessage(), x);
+ throw new URISyntaxException(url,x.getMessage());
+ } catch (Exception x) {
+
+ getLog().warn(x.getMessage(), x);
+ throw new Exception(x.getMessage());
+ }
+ }
+
+ private MRSimplerBatchPublisher ( Collection<String> hosts, String topic, int maxBatchSize, long maxBatchAgeMs, boolean compress) throws MalformedURLException
+ {
+ super ( hosts );
+
+ if ( topic == null || topic.length() < 1 )
+ {
+ throw new IllegalArgumentException ( "A topic must be provided." );
+ }
+
+ fHostSelector = new HostSelector(hosts, null);
+ fClosed = false;
+ fTopic = topic;
+ fMaxBatchSize = maxBatchSize;
+ fMaxBatchAgeMs = maxBatchAgeMs;
+ fCompress = compress;
+
+ fPending = new LinkedBlockingQueue<TimestampedMessage> ();
+ fDontSendUntilMs = 0;
+ fExec = new ScheduledThreadPoolExecutor ( 1 );
+ pubResponse = new MRPublisherResponse();
+
+ }
+
+ private MRSimplerBatchPublisher ( Collection<String> hosts, String topic, int maxBatchSize, long maxBatchAgeMs, boolean compress, boolean allowSelfSignedCerts, int httpThreadOccurnace ) throws MalformedURLException
+ {
+ super ( hosts );
+
+ if ( topic == null || topic.length() < 1 )
+ {
+ throw new IllegalArgumentException ( "A topic must be provided." );
+ }
+
+ fHostSelector = new HostSelector(hosts, null);
+ fClosed = false;
+ fTopic = topic;
+ fMaxBatchSize = maxBatchSize;
+ fMaxBatchAgeMs = maxBatchAgeMs;
+ fCompress = compress;
+ threadOccuranceTime=httpThreadOccurnace;
+ fPending = new LinkedBlockingQueue<TimestampedMessage> ();
+ fDontSendUntilMs = 0;
+ fExec = new ScheduledThreadPoolExecutor ( 1 );
+ fExec.scheduleAtFixedRate ( new Runnable()
+ {
+ @Override
+ public void run ()
+ {
+ send ( false );
+ }
+ }, 100, threadOccuranceTime, TimeUnit.MILLISECONDS );
+ }
+
+ private static class TimestampedMessage extends message
+ {
+ public TimestampedMessage ( message m )
+ {
+ super ( m );
+ timestamp = Clock.now();
+ }
+ public final long timestamp;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public void setUsername(String username) {
+ this.username = username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public void setHost(String host) {
+ this.host = host;
+ }
+
+ public String getContentType() {
+ return contentType;
+ }
+
+ public void setContentType(String contentType) {
+ this.contentType = contentType;
+ }
+
+ public String getAuthKey() {
+ return authKey;
+ }
+
+ public void setAuthKey(String authKey) {
+ this.authKey = authKey;
+ }
+
+ public String getAuthDate() {
+ return authDate;
+ }
+
+ public void setAuthDate(String authDate) {
+ this.authDate = authDate;
+ }
+
+}
diff --git a/src/main/java/com/att/nsa/mr/client/response/MRConsumerResponse.java b/src/main/java/com/att/nsa/mr/client/response/MRConsumerResponse.java
new file mode 100644
index 0000000..102794e
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/response/MRConsumerResponse.java
@@ -0,0 +1,60 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client.response;
+
+public class MRConsumerResponse {
+
+ private String responseCode;
+
+ private String responseMessage;
+
+ private Iterable<String> actualMessages;
+
+
+
+
+ public String getResponseCode() {
+ return responseCode;
+ }
+
+ public void setResponseCode(String responseCode) {
+ this.responseCode = responseCode;
+ }
+
+ public String getResponseMessage() {
+ return responseMessage;
+ }
+
+ public void setResponseMessage(String responseMessage) {
+ this.responseMessage = responseMessage;
+ }
+
+ public Iterable<String> getActualMessages() {
+ return actualMessages;
+ }
+
+ public void setActualMessages(Iterable<String> actualMessages) {
+ this.actualMessages = actualMessages;
+ }
+
+
+}
diff --git a/src/main/java/com/att/nsa/mr/client/response/MRPublisherResponse.java b/src/main/java/com/att/nsa/mr/client/response/MRPublisherResponse.java
new file mode 100644
index 0000000..4aebe23
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/response/MRPublisherResponse.java
@@ -0,0 +1,66 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.client.response;
+
+/**
+ * Response for Publisher
+ * @author author
+ *
+ */
+public class MRPublisherResponse {
+ private String responseCode;
+
+ private String responseMessage;
+
+ private int pendingMsgs;
+
+ public String getResponseCode() {
+ return responseCode;
+ }
+
+ public void setResponseCode(String responseCode) {
+ this.responseCode = responseCode;
+ }
+
+ public String getResponseMessage() {
+ return responseMessage;
+ }
+
+ public void setResponseMessage(String responseMessage) {
+ this.responseMessage = responseMessage;
+ }
+
+ public int getPendingMsgs() {
+ return pendingMsgs;
+ }
+
+ public void setPendingMsgs(int pendingMsgs) {
+ this.pendingMsgs = pendingMsgs;
+ }
+
+ public String toString() {
+ return "Response Code:" + this.responseCode + ","
+ + "Response Message:" + this.responseMessage + "," + "Pending Messages Count"
+ + this.pendingMsgs;
+ }
+
+}
diff --git a/src/main/java/com/att/nsa/mr/dme/client/DefaultLoggingFailoverFaultHandler.java b/src/main/java/com/att/nsa/mr/dme/client/DefaultLoggingFailoverFaultHandler.java
new file mode 100644
index 0000000..1d3b6be
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/dme/client/DefaultLoggingFailoverFaultHandler.java
@@ -0,0 +1,50 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.dme.client;
+
+//package com.att.aft.dme2.api;
+
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+//import com.att.aft.dme2.api.DME2FailoverFaultHandler;
+//import com.att.aft.dme2.api.util.DME2Constants;
+//import com.att.aft.dme2.api.util.DME2ExchangeFaultContext;
+//import com.att.aft.dme2.api.util.LogMessage;
+//import com.att.aft.dme2.api.util.LogUtil;
+public class DefaultLoggingFailoverFaultHandler /*implements DME2FailoverFaultHandler*/ {
+ /** The logger. */
+ //private static Logger logger = DME2Constants.getLogger(DefaultLoggingFailoverFaultHandler.class.getName());
+
+// @Override
+// public void handleEndpointFailover(/*DME2ExchangeFaultContext context*/) {
+// // LogUtil.INSTANCE.report(logger, Level.WARNING, LogMessage.SEP_FAILOVER, context.getService(),context.getRequestURL(),context.getRouteOffer(),context.getResponseCode(),context.getException());
+// }
+// @Override
+// /**
+// * The DME2Exchange already has a log message when the route offer is failed over. We dont need to log it again here.
+// */
+// public void handleRouteOfferFailover(DME2ExchangeFaultContext context) {
+// //noop
+//
+// }
+} \ No newline at end of file
diff --git a/src/main/java/com/att/nsa/mr/dme/client/HeaderReplyHandler.java b/src/main/java/com/att/nsa/mr/dme/client/HeaderReplyHandler.java
new file mode 100644
index 0000000..c70c63f
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/dme/client/HeaderReplyHandler.java
@@ -0,0 +1,63 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.dme.client;
+
+
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.aft.dme2.api.util.DME2ExchangeFaultContext;
+import com.att.aft.dme2.api.util.DME2ExchangeReplyHandler;
+import com.att.aft.dme2.api.util.DME2ExchangeResponseContext;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.impl.MRSimplerBatchPublisher;
+
+
+
+//public class HeaderReplyHandler implements DME2ReplyHandler {
+
+ public class HeaderReplyHandler implements DME2ExchangeReplyHandler {
+
+ private Logger fLog = LoggerFactory.getLogger ( this.getClass().getName () );
+
+
+ @Override public void handleFault(DME2ExchangeFaultContext responseData) {
+ // TODO Auto-generated method stub
+ //StaticCache.getInstance().setHandleFaultInvoked(true);
+ }
+ @Override public void handleEndpointFault(DME2ExchangeFaultContext responseData) {
+ // TODO Auto-generated method stub
+ //StaticCache.getInstance().setHandleEndpointFaultInvoked(true);
+ }
+@Override public void handleReply(DME2ExchangeResponseContext responseData) {
+
+ if(responseData != null) {
+ MRClientFactory.DME2HeadersMap=responseData.getResponseHeaders();
+ if (responseData.getResponseHeaders().get("transactionId")!=null)
+ fLog.info("Transaction Id : " + responseData.getResponseHeaders().get("transactionId"));
+
+ }
+}
+
+}
diff --git a/src/main/java/com/att/nsa/mr/dme/client/PreferredRouteReplyHandler.java b/src/main/java/com/att/nsa/mr/dme/client/PreferredRouteReplyHandler.java
new file mode 100644
index 0000000..b7e174c
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/dme/client/PreferredRouteReplyHandler.java
@@ -0,0 +1,74 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.dme.client;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.InputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.aft.dme2.api.util.DME2ExchangeFaultContext;
+import com.att.aft.dme2.api.util.DME2ExchangeReplyHandler;
+import com.att.aft.dme2.api.util.DME2ExchangeResponseContext;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.impl.MRSimplerBatchPublisher;
+
+public class PreferredRouteReplyHandler implements DME2ExchangeReplyHandler {
+ private Logger fLog = LoggerFactory.getLogger ( this.getClass().getName () );
+ @Override public void handleReply(DME2ExchangeResponseContext responseData) {
+
+ if(responseData != null) {
+ MRClientFactory.DME2HeadersMap=responseData.getResponseHeaders();
+ if (responseData.getResponseHeaders().get("transactionId")!=null)
+
+ fLog.info("Transaction_Id : " + responseData.getResponseHeaders().get("transactionId"));
+
+ if(responseData.getRouteOffer() != null ){
+ routeWriter("preferredRouteKey",responseData.getRouteOffer());
+
+ }
+ }
+}
+
+ @Override public void handleFault(DME2ExchangeFaultContext responseData) {
+ // TODO Auto-generated method stub
+ //StaticCache.getInstance().setHandleFaultInvoked(true);
+ }
+ @Override public void handleEndpointFault(DME2ExchangeFaultContext responseData) {
+ // TODO Auto-generated method stub
+ //StaticCache.getInstance().setHandleEndpointFaultInvoked(true);
+ }
+ public void routeWriter(String routeKey, String routeValue){
+
+ try{
+
+ FileWriter routeWriter=new FileWriter(new File (MRSimplerBatchPublisher.routerFilePath));
+ routeWriter.write(routeKey+"="+routeValue);
+ routeWriter.close();
+
+ }catch(Exception ex){
+ fLog.error("Reply Router Error " + ex.toString() );
+ }
+
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/dme/client/PreferredRouteRequestHandler.java b/src/main/java/com/att/nsa/mr/dme/client/PreferredRouteRequestHandler.java
new file mode 100644
index 0000000..a65683b
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/dme/client/PreferredRouteRequestHandler.java
@@ -0,0 +1,54 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.dme.client;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.aft.dme2.api.util.DME2ExchangeRequestContext;
+import com.att.aft.dme2.api.util.DME2ExchangeRequestHandler;
+import com.att.nsa.mr.client.MRClientFactory;
+
+public class PreferredRouteRequestHandler implements DME2ExchangeRequestHandler {
+ private Logger fLog = LoggerFactory.getLogger(this.getClass().getName());
+
+ @Override
+ public void handleRequest(DME2ExchangeRequestContext requestData) {
+
+ if (requestData != null) {
+
+ requestData.setPreferredRouteOffer(readRoute("preferredRouteKey"));
+ }
+ }
+
+ public String readRoute(String routeKey) {
+
+ try {
+
+ MRClientFactory.prop.load(MRClientFactory.routeReader);
+
+ } catch (Exception ex) {
+ fLog.error("Request Router Error " + ex.toString());
+ }
+ return MRClientFactory.prop.getProperty(routeKey);
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/dme/client/SimpleExampleConsumer.java b/src/main/java/com/att/nsa/mr/dme/client/SimpleExampleConsumer.java
new file mode 100644
index 0000000..4df9fcb
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/dme/client/SimpleExampleConsumer.java
@@ -0,0 +1,77 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.dme.client;
+
+import java.util.Map;
+
+import javax.ws.rs.core.MultivaluedMap;
+
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRConsumer;
+
+public class SimpleExampleConsumer {
+
+ public static void main(String[] args) {
+
+ long count = 0;
+ long nextReport = 5000;
+
+ final long startMs = System.currentTimeMillis();
+
+ try {
+
+ final MRConsumer cc = MRClientFactory.createConsumer("D:\\SG\\consumer.properties");
+ while (true) {
+ for (String msg : cc.fetch()) {
+
+ System.out.println("Message Received: " + msg);
+ }
+ // Header for DME2 Call.
+ MultivaluedMap<String, Object> headersMap = MRClientFactory.HTTPHeadersMap;
+ for (String key : headersMap.keySet()) {
+ System.out.println("Header Key " + key);
+ System.out.println("Header Value " + headersMap.get(key));
+ }
+ // Header for HTTP Call.
+
+ Map<String, String>
+ dme2headersMap=MRClientFactory.DME2HeadersMap; for(String key
+ : dme2headersMap.keySet()) { System.out.println("Header Key "
+ + key); System.out.println("Header Value " +
+ dme2headersMap.get(key)); }
+
+ if (count > nextReport) {
+ nextReport += 5000;
+
+ final long endMs = System.currentTimeMillis();
+ final long elapsedMs = endMs - startMs;
+ final double elapsedSec = elapsedMs / 1000.0;
+ final double eps = count / elapsedSec;
+ System.out.println("Consumed " + count + " in " + elapsedSec + "; " + eps + " eps");
+ }
+ }
+ } catch (Exception x) {
+ System.err.println(x.getClass().getName() + ": " + x.getMessage());
+ }
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/dme/client/SimpleExamplePublisher.java b/src/main/java/com/att/nsa/mr/dme/client/SimpleExamplePublisher.java
new file mode 100644
index 0000000..0e8efda
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/dme/client/SimpleExamplePublisher.java
@@ -0,0 +1,135 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.dme.client;
+
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import java.util.concurrent.TimeUnit;
+
+import javax.ws.rs.core.MultivaluedMap;
+
+import org.json.JSONObject;
+
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRPublisher.message;
+
+/**
+ * An example of how to use the Java publisher.
+ *
+ * @author author
+ */
+public class SimpleExamplePublisher {
+ static String content = null;
+ static String messageSize = null;
+ static String transport = null;
+ static String messageCount = null;
+
+ public void publishMessage(String producerFilePath) throws IOException, InterruptedException, Exception {
+
+ // create our publisher
+
+ // publish some messages
+
+
+ StringBuilder sb = new StringBuilder();
+ final MRBatchingPublisher pub = MRClientFactory.createBatchingPublisher(producerFilePath);
+
+ if (content.equalsIgnoreCase("text/plain")) {
+ for (int i = 0; i < Integer.parseInt(messageCount); i++) {
+ for (int j = 0; j < Integer.parseInt(messageSize); j++) {
+ sb.append("T");
+ }
+
+ pub.send(sb.toString());
+ }
+ } else if (content.equalsIgnoreCase("application/cambria")) {
+ for (int i = 0; i < Integer.parseInt(messageCount); i++) {
+ for (int j = 0; j < Integer.parseInt(messageSize); j++) {
+ sb.append("C");
+ }
+
+ pub.send("Key", sb.toString());
+ }
+ } else if (content.equalsIgnoreCase("application/json")) {
+ for (int i = 0; i < Integer.parseInt(messageCount); i++) {
+
+ final JSONObject msg12 = new JSONObject();
+ msg12.put("Name", "DMaaP Reference Client to Test jason Message");
+
+ pub.send(msg12.toString());
+
+ }
+ }
+
+ // ...
+
+ // close the publisher to make sure everything's sent before exiting.
+ // The batching
+ // publisher interface allows the app to get the set of unsent messages.
+ // It could
+ // write them to disk, for example, to try to send them later.
+ /* final List<message> stuck = pub.close(20, TimeUnit.SECONDS);
+ if (stuck.size() > 0) {
+ System.err.println(stuck.size() + " messages unsent");
+ } else {
+ System.out.println("Clean exit; all messages sent.");
+ }*/
+
+ if (transport.equalsIgnoreCase("HTTP")) {
+ MultivaluedMap<String, Object> headersMap = MRClientFactory.HTTPHeadersMap;
+ for (String key : headersMap.keySet()) {
+ System.out.println("Header Key " + key);
+ System.out.println("Header Value " + headersMap.get(key));
+ }
+ } else {
+ Map<String, String> dme2headersMap = MRClientFactory.DME2HeadersMap;
+ for (String key : dme2headersMap.keySet()) {
+ System.out.println("Header Key " + key);
+ System.out.println("Header Value " + dme2headersMap.get(key));
+ }
+ }
+
+ }
+
+ public static void main(String[] args) throws InterruptedException, Exception {
+
+ String producerFilePath = args[0];
+ content = args[1];
+ messageSize = args[2];
+ transport = args[3];
+ messageCount = args[4];
+ /*String producerFilePath = null;
+ content = null;
+ messageSize =null;
+ transport =null;
+ messageCount = null;*/
+ SimpleExamplePublisher publisher = new SimpleExamplePublisher();
+
+ publisher.publishMessage("D:\\SG\\producer.properties");
+ }
+
+}
diff --git a/src/main/java/com/att/nsa/mr/logging/MRAppender.java b/src/main/java/com/att/nsa/mr/logging/MRAppender.java
new file mode 100644
index 0000000..8faf0df
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/logging/MRAppender.java
@@ -0,0 +1,160 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.logging;
+
+import java.io.IOException;
+
+import org.apache.log4j.AppenderSkeleton;
+import org.apache.log4j.helpers.LogLog;
+import org.apache.log4j.spi.LoggingEvent;
+
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRPublisher;
+
+/**
+ * @author author
+ *
+ */
+public class MRAppender extends AppenderSkeleton {
+
+ private MRPublisher fPublisher;
+
+ //Provided through log4j configuration
+ private String topic;
+ private String partition;
+ private String hosts;
+ private int maxBatchSize = 1;
+ private int maxAgeMs = 1000;
+ private boolean compress = false;
+
+ /**
+ *
+ */
+ public MRAppender() {
+ super();
+ }
+
+ /**
+ * @param isActive
+ */
+ public MRAppender(boolean isActive) {
+ super(isActive);
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.log4j.Appender#close()
+ */
+ @Override
+ public void close() {
+ if (!this.closed) {
+ this.closed = true;
+ fPublisher.close();
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.log4j.Appender#requiresLayout()
+ */
+ @Override
+ public boolean requiresLayout() {
+ return false;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.log4j.AppenderSkeleton#append(org.apache.log4j.spi.LoggingEvent)
+ */
+ @Override
+ protected void append(LoggingEvent event) {
+ final String message;
+
+ if (this.layout == null) {
+ message = event.getRenderedMessage();
+ } else {
+ message = this.layout.format(event);
+ }
+
+ try {
+ fPublisher.send(partition, message);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ public void activateOptions() {
+ if (hosts != null && topic != null && partition != null) {
+ fPublisher = MRClientFactory.createBatchingPublisher(hosts.split(","), topic, maxBatchSize, maxAgeMs, compress);
+ } else {
+ LogLog.error("The Hosts, Topic, and Partition parameter are required to create a MR Log4J Appender");
+ }
+ }
+ public String getTopic() {
+ return topic;
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ public String getPartition() {
+ return partition;
+ }
+
+ public void setPartition(String partition) {
+ this.partition = partition;
+ }
+
+ public String getHosts() {
+ return hosts;
+ }
+
+ public void setHosts(String hosts) {
+ this.hosts = hosts;
+ }
+
+ public int getMaxBatchSize() {
+ return maxBatchSize;
+ }
+
+ public void setMaxBatchSize(int maxBatchSize) {
+ this.maxBatchSize = maxBatchSize;
+ }
+
+ public int getMaxAgeMs() {
+ return maxAgeMs;
+ }
+
+ public void setMaxAgeMs(int maxAgeMs) {
+ this.maxAgeMs = maxAgeMs;
+ }
+
+ public boolean isCompress() {
+ return compress;
+ }
+
+ public void setCompress(boolean compress) {
+ this.compress = compress;
+ }
+
+}
diff --git a/src/main/java/com/att/nsa/mr/test/clients/ConsolePublisher.java b/src/main/java/com/att/nsa/mr/test/clients/ConsolePublisher.java
new file mode 100644
index 0000000..2294d7b
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/test/clients/ConsolePublisher.java
@@ -0,0 +1,87 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.test.clients;
+
+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;
+
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRPublisher.message;
+
+/**
+ * A simple publisher that reads from std in, sending each line as a message.
+ * @author author
+ */
+public class 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] : "aaa.it.att.com,bbb.it.att.com,ccc.it.att.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 )
+ {
+ System.err.println ( "Send on close interrupted." );
+ }
+ for ( message m : leftovers )
+ {
+ System.err.println ( "Unsent message: " + m.fMsg );
+ }
+ }
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/test/clients/ProtocolTypeConstants.java b/src/main/java/com/att/nsa/mr/test/clients/ProtocolTypeConstants.java
new file mode 100644
index 0000000..6e86d47
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/test/clients/ProtocolTypeConstants.java
@@ -0,0 +1,46 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.test.clients;
+
+/**
+ * @author author
+ *
+ */
+public enum ProtocolTypeConstants {
+
+ DME2("DME2"),
+ AAF_AUTH("HTTPAAF"),
+ AUTH_KEY("HTTPAUTH");
+
+ private String value;
+
+ private ProtocolTypeConstants(String value) {
+ this.value = value;
+ }
+
+ public String getValue() {
+ return value;
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/test/clients/SampleConsumer.java b/src/main/java/com/att/nsa/mr/test/clients/SampleConsumer.java
new file mode 100644
index 0000000..8e1c0e0
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/test/clients/SampleConsumer.java
@@ -0,0 +1,87 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.test.clients;
+
+import java.io.IOException;
+import java.util.LinkedList;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRConsumer;
+
+public class SampleConsumer {
+ public static void main ( String[] args )
+ {
+ final Logger LOG = LoggerFactory.getLogger(SampleConsumer.class);
+
+
+ LOG.info("Sample Consumer Class executing");
+ final String topic = "com.att.app.dmaap.mr.testingTopic";
+ final String url = ( args.length > 1 ? args[1] : "localhost:8181" );
+ final String group = ( args.length > 2 ? args[2] :"grp" );
+ /*final String id = ( args.length > 3 ? args[3] : "0" );*/
+ final String id = ( args.length > 3 ? args[3] : "1" );
+
+ long count = 0;
+ long nextReport = 5000;
+
+ final long startMs = System.currentTimeMillis ();
+
+ final LinkedList<String> urlList = new LinkedList<String> ();
+ for ( String u : url.split ( "," ) )
+ {
+ urlList.add ( u );
+ }
+
+ final MRConsumer cc = MRClientFactory.createConsumer ( urlList, topic, group, id, 10*1000, 1000, null, "CG0TXc2Aa3v8LfBk", "pj2rhxJWKP23pgy8ahMnjH88" );
+ try
+ {
+ while ( true )
+ {
+ for ( String msg : cc.fetch () )
+ {
+ //System.out.println ( "" + (++count) + ": " + msg );
+ LOG.info ( "" + (++count) + ": " + msg );
+ }
+
+ if ( count > nextReport )
+ {
+ nextReport += 5000;
+
+ final long endMs = System.currentTimeMillis ();
+ final long elapsedMs = endMs - startMs;
+ final double elapsedSec = elapsedMs / 1000.0;
+ final double eps = count / elapsedSec;
+ //System.out.println ( "Consumed " + count + " in " + elapsedSec + "; " + eps + " eps" );
+ LOG.info ( "Consumed " + count + " in " + elapsedSec + "; " + eps + " eps" );
+ }
+ LOG.info ( "" + (++count) + ": consumed message" );
+ }
+ }
+ catch ( Exception x )
+ {
+ System.err.println ( x.getClass().getName () + ": " + x.getMessage () );
+ }
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/test/clients/SamplePublisher.java b/src/main/java/com/att/nsa/mr/test/clients/SamplePublisher.java
new file mode 100644
index 0000000..36c2b7f
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/test/clients/SamplePublisher.java
@@ -0,0 +1,86 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.test.clients;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.json.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRClientBuilders.PublisherBuilder;
+import com.att.nsa.mr.client.MRPublisher.message;
+
+public class SamplePublisher {
+ public static void main ( String[] args ) throws IOException, InterruptedException
+ {
+ final Logger LOG = LoggerFactory.getLogger(SampleConsumer.class);
+ // read the hosts(s) from the command line
+ final String hosts = ( args.length > 0 ? args[0] : "localhost:8181" );
+
+ // read the topic name from the command line
+ //final String topic = ( args.length > 1 ? args[1] : "MY-EXAMPLE-TOPIC" );
+ final String topic = ( args.length > 1 ? args[1] : "com.att.app.dmaap.mr.testingTopic" );
+
+ // set up some batch limits and the compression flag
+ final int maxBatchSize = 100;
+ final int maxAgeMs = 250;
+ final boolean withGzip = false;
+
+ // create our publisher
+
+ final MRBatchingPublisher pub = new PublisherBuilder ().
+ usingHosts ( hosts ).
+ onTopic ( topic ).limitBatch(maxBatchSize, maxAgeMs).
+ authenticatedBy ( "CG0TXc2Aa3v8LfBk", "pj2rhxJWKP23pgy8ahMnjH88" ).
+ build ()
+ ;
+ // publish some messages
+ final JSONObject msg1 = new JSONObject ();
+ msg1.put ( "name", "tttttttttttttttt" );
+ msg1.put ( "greeting", "ooooooooooooooooo" );
+ pub.send ( "MyPartitionKey", msg1.toString () );
+
+ final JSONObject msg2 = new JSONObject ();
+ msg2.put ( "now", System.currentTimeMillis () );
+ pub.send ( "MyOtherPartitionKey", msg2.toString () );
+
+ // ...
+
+ // close the publisher to make sure everything's sent before exiting. The batching
+ // publisher interface allows the app to get the set of unsent messages. It could
+ // write them to disk, for example, to try to send them later.
+ final List<message> stuck = pub.close ( 20, TimeUnit.SECONDS );
+ if ( stuck.size () > 0 )
+ {
+ LOG.warn ( stuck.size() + " messages unsent" );
+ }
+ else
+ {
+ LOG.info ( "Clean exit; all messages sent." );
+ }
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/test/clients/SimpleExampleConsumer.java b/src/main/java/com/att/nsa/mr/test/clients/SimpleExampleConsumer.java
new file mode 100644
index 0000000..3a131a0
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/test/clients/SimpleExampleConsumer.java
@@ -0,0 +1,86 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.test.clients;
+
+
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.util.Properties;
+
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRConsumer;
+
+public class SimpleExampleConsumer
+{
+
+ static FileWriter routeWriter= null;
+ static Properties props=null;
+ static FileReader routeReader=null;
+ public static void main ( String[] args )
+ {
+
+ long count = 0;
+ long nextReport = 5000;
+
+ final long startMs = System.currentTimeMillis ();
+
+ try
+ {
+ String routeFilePath="/src/main/resources/dme2/preferredRoute.txt";
+
+
+ File fo= new File(routeFilePath);
+ if(!fo.exists()){
+ routeWriter=new FileWriter(new File (routeFilePath));
+ }
+ routeReader= new FileReader(new File (routeFilePath));
+ props= new Properties();
+ final MRConsumer cc = MRClientFactory.createConsumer ( "/src/main/resources/dme2/consumer.properties" );
+ while ( true )
+ {
+ for ( String msg : cc.fetch () )
+ {
+ //System.out.println ( "" + (++count) + ": " + msg );
+ System.out.println(msg);
+ }
+
+ if ( count > nextReport )
+ {
+ nextReport += 5000;
+
+ final long endMs = System.currentTimeMillis ();
+ final long elapsedMs = endMs - startMs;
+ final double elapsedSec = elapsedMs / 1000.0;
+ final double eps = count / elapsedSec;
+ System.out.println ( "Consumed " + count + " in " + elapsedSec + "; " + eps + " eps" );
+ }
+ }
+ }
+ catch ( Exception x )
+ {
+ System.err.println ( x.getClass().getName () + ": " + x.getMessage () );
+ }
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/test/clients/SimpleExampleConsumerWithReturnResponse.java b/src/main/java/com/att/nsa/mr/test/clients/SimpleExampleConsumerWithReturnResponse.java
new file mode 100644
index 0000000..b3c167b
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/test/clients/SimpleExampleConsumerWithReturnResponse.java
@@ -0,0 +1,90 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.test.clients;
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.util.Properties;
+
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRConsumer;
+import com.att.nsa.mr.client.response.MRConsumerResponse;
+
+public class SimpleExampleConsumerWithReturnResponse {
+
+
+ static FileWriter routeWriter= null;
+ static Properties props=null;
+ static FileReader routeReader=null;
+ public static void main ( String[] args )
+ {
+
+ long count = 0;
+ long nextReport = 5000;
+
+ final long startMs = System.currentTimeMillis ();
+
+ try
+ {
+ String routeFilePath="src/main/resources/dme2/preferredRoute.txt";
+
+
+ File fo= new File(routeFilePath);
+ if(!fo.exists()){
+ routeWriter=new FileWriter(new File (routeFilePath));
+ }
+ routeReader= new FileReader(new File (routeFilePath));
+ props= new Properties();
+ final MRConsumer cc = MRClientFactory.createConsumer ( "src/main/resources/dme2/consumer.properties" );
+ while ( true )
+ {
+ MRConsumerResponse mrConsumerResponse = cc.fetchWithReturnConsumerResponse();
+ System.out.println("mrConsumerResponse code :"+mrConsumerResponse.getResponseCode());
+
+ System.out.println("mrConsumerResponse Message :"+mrConsumerResponse.getResponseMessage());
+
+ System.out.println("mrConsumerResponse ActualMessage :"+mrConsumerResponse.getActualMessages());
+ /*for ( String msg : mrConsumerResponse.getActualMessages() )
+ {
+ //System.out.println ( "" + (++count) + ": " + msg );
+ System.out.println(msg);
+ }*/
+ if ( count > nextReport )
+ {
+ nextReport += 5000;
+
+ final long endMs = System.currentTimeMillis ();
+ final long elapsedMs = endMs - startMs;
+ final double elapsedSec = elapsedMs / 1000.0;
+ final double eps = count / elapsedSec;
+ System.out.println ( "Consumed " + count + " in " + elapsedSec + "; " + eps + " eps" );
+ }
+ }
+ }
+ catch ( Exception x )
+ {
+ System.err.println ( x.getClass().getName () + ": " + x.getMessage () );
+ }
+ }
+
+}
diff --git a/src/main/java/com/att/nsa/mr/test/clients/SimpleExamplePublisher.java b/src/main/java/com/att/nsa/mr/test/clients/SimpleExamplePublisher.java
new file mode 100644
index 0000000..01e770e
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/test/clients/SimpleExamplePublisher.java
@@ -0,0 +1,98 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.test.clients;
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.json.JSONObject;
+
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRPublisher.message;
+
+/**
+ * An example of how to use the Java publisher.
+ * @author author
+ */
+public class SimpleExamplePublisher
+{
+ static FileWriter routeWriter= null;
+ static Properties props=null;
+ static FileReader routeReader=null;
+ public void publishMessage ( String producerFilePath ) throws IOException, InterruptedException, Exception
+ {
+
+ // create our publisher
+ final MRBatchingPublisher pub = MRClientFactory.createBatchingPublisher (producerFilePath);
+ // publish some messages
+ final JSONObject msg1 = new JSONObject ();
+ msg1.put ( "Name", "Sprint" );
+ //msg1.put ( "greeting", "Hello .." );
+ pub.send ( "First cambria messge" );
+ pub.send ( "MyPartitionKey", msg1.toString () );
+
+ final JSONObject msg2 = new JSONObject ();
+ //msg2.put ( "mrclient1", System.currentTimeMillis () );
+
+
+ // ...
+
+ // close the publisher to make sure everything's sent before exiting. The batching
+ // publisher interface allows the app to get the set of unsent messages. It could
+ // write them to disk, for example, to try to send them later.
+ final List<message> stuck = pub.close ( 20, TimeUnit.SECONDS );
+ if ( stuck.size () > 0 )
+ {
+ System.err.println ( stuck.size() + " messages unsent" );
+ }
+ else
+ {
+ System.out.println ( "Clean exit; all messages sent." );
+ }
+ }
+
+ public static void main(String []args) throws InterruptedException, Exception{
+
+ String routeFilePath="/src/main/resources/dme2/preferredRoute.txt";
+
+ SimpleExamplePublisher publisher = new SimpleExamplePublisher();
+
+
+ File fo= new File(routeFilePath);
+ if(!fo.exists()){
+ routeWriter=new FileWriter(new File (routeFilePath));
+ }
+ routeReader= new FileReader(new File (routeFilePath));
+ props= new Properties();
+ publisher.publishMessage("/src/main/resources/dme2/producer.properties");
+ }
+
+}
+
diff --git a/src/main/java/com/att/nsa/mr/test/clients/SimpleExamplePublisherWithResponse.java b/src/main/java/com/att/nsa/mr/test/clients/SimpleExamplePublisherWithResponse.java
new file mode 100644
index 0000000..4914688
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/test/clients/SimpleExamplePublisherWithResponse.java
@@ -0,0 +1,84 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.test.clients;
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import org.json.JSONObject;
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.response.MRPublisherResponse;
+ /**
+ *An example of how to use the Java publisher.
+ * @author author
+ *
+ */
+ public class SimpleExamplePublisherWithResponse
+ {
+ static FileWriter routeWriter= null;
+ static Properties props=null;
+ static FileReader routeReader=null;
+
+ public static void main(String []args) throws InterruptedException, Exception{
+
+ String routeFilePath="src/main/resources/dme2/preferredRoute.txt";
+ String msgCount = args[0];
+ SimpleExamplePublisherWithResponse publisher = new SimpleExamplePublisherWithResponse();
+ File fo= new File(routeFilePath);
+ if(!fo.exists()){
+ routeWriter=new FileWriter(new File (routeFilePath));
+ }
+ routeReader= new FileReader(new File (routeFilePath));
+ props= new Properties();
+ int i=0;
+ while (i< Integer.valueOf(msgCount))
+ {
+ publisher.publishMessage("src/main/resources/dme2/producer.properties",Integer.valueOf(msgCount));
+ i++;
+ }
+ }
+
+ public void publishMessage ( String producerFilePath , int count ) throws IOException, InterruptedException, Exception
+ {
+ // create our publisher
+ final MRBatchingPublisher pub = MRClientFactory.createBatchingPublisher (producerFilePath,true);
+ // publish some messages
+ final JSONObject msg1 = new JSONObject ();
+
+ msg1.put ( "Partition:1", "Message:"+count);
+ msg1.put ( "greeting", "Hello .." );
+
+
+ pub.send ( "1", msg1.toString());
+ pub.send ( "1", msg1.toString());
+
+ MRPublisherResponse res= pub.sendBatchWithResponse();
+
+ System.out.println("Pub response->"+res.toString());
+ }
+
+
+ }
diff --git a/src/main/java/com/att/nsa/mr/test/support/MRBatchingPublisherMock.java b/src/main/java/com/att/nsa/mr/test/support/MRBatchingPublisherMock.java
new file mode 100644
index 0000000..3c123b1
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/test/support/MRBatchingPublisherMock.java
@@ -0,0 +1,184 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.test.support;
+
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.response.MRPublisherResponse;
+
+/**
+ * A helper for unit testing systems that use a MRPublisher. When setting
+ * up your test, inject an instance into MRClientFactory to have it return
+ * the mock client.
+ *
+ * @author author
+ *
+ */
+public class MRBatchingPublisherMock implements MRBatchingPublisher
+{
+ public class Entry
+ {
+ public Entry ( String partition, String msg )
+ {
+ fPartition = partition;
+ fMessage = msg;
+ }
+
+ @Override
+ public String toString ()
+ {
+ return fMessage;
+ }
+
+ public final String fPartition;
+ public final String fMessage;
+ }
+
+ public MRBatchingPublisherMock ()
+ {
+ fCaptures = new LinkedList<Entry> ();
+ }
+
+ public interface Listener
+ {
+ void onMessage ( Entry e );
+ }
+ public void addListener ( Listener listener )
+ {
+ fListeners.add ( listener );
+ }
+
+ public List<Entry> getCaptures ()
+ {
+ return getCaptures ( new MessageFilter () { @Override public boolean match ( String msg ) { return true; } } );
+ }
+
+ public interface MessageFilter
+ {
+ boolean match ( String msg );
+ }
+
+ public List<Entry> getCaptures ( MessageFilter filter )
+ {
+ final LinkedList<Entry> result = new LinkedList<Entry> ();
+ for ( Entry capture : fCaptures )
+ {
+ if ( filter.match ( capture.fMessage ) )
+ {
+ result.add ( capture );
+ }
+ }
+ return result;
+ }
+
+ public int received ()
+ {
+ return fCaptures.size();
+ }
+
+ public void reset ()
+ {
+ fCaptures.clear ();
+ }
+
+ @Override
+ public int send ( String partition, String msg )
+ {
+ final Entry e = new Entry ( partition, msg );
+
+ fCaptures.add ( e );
+ for ( Listener l : fListeners )
+ {
+ l.onMessage ( e );
+ }
+ return 1;
+ }
+
+ @Override
+ public int send ( message msg )
+ {
+ return send ( msg.fPartition, msg.fMsg );
+ }
+ @Override
+ public int send ( String msg )
+ {
+ return 1;
+
+ }
+
+ @Override
+ public int send ( Collection<message> msgs )
+ {
+ int sum = 0;
+ for ( message m : msgs )
+ {
+ sum += send ( m );
+ }
+ return sum;
+ }
+
+ @Override
+ public int getPendingMessageCount ()
+ {
+ return 0;
+ }
+
+ @Override
+ public List<message> close ( long timeout, TimeUnit timeoutUnits )
+ {
+ return new LinkedList<message> ();
+ }
+
+ @Override
+ public void close ()
+ {
+ }
+
+ @Override
+ public void setApiCredentials ( String apiKey, String apiSecret )
+ {
+ }
+
+ @Override
+ public void clearApiCredentials ()
+ {
+ }
+
+ @Override
+ public void logTo ( Logger log )
+ {
+ }
+
+ private final LinkedList<Entry> fCaptures;
+ private LinkedList<Listener> fListeners = new LinkedList<Listener> ();
+ @Override
+ public MRPublisherResponse sendBatchWithResponse() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/test/support/MRConsumerMock.java b/src/main/java/com/att/nsa/mr/test/support/MRConsumerMock.java
new file mode 100644
index 0000000..3373710
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/test/support/MRConsumerMock.java
@@ -0,0 +1,168 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.test.support;
+
+import java.io.IOException;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.slf4j.Logger;
+
+import com.att.nsa.mr.client.MRConsumer;
+import com.att.nsa.mr.client.response.MRConsumerResponse;
+
+/**
+ * A helper for unit testing systems that use a MRConsumer. When setting
+ * up your test, inject an instance into MRClientFactory to have it return
+ * the mock client.
+ *
+ * @author author
+ *
+ */
+public class MRConsumerMock implements MRConsumer
+{
+ public class Entry
+ {
+ public Entry ( long waitMs, int statusCode, List<String> msgs )
+ {
+ fWaitMs = waitMs;
+ fStatusCode = statusCode;
+ fStatusMsg = null;
+ fMsgs = new LinkedList<String> ( msgs );
+ }
+
+ public Entry ( long waitMs, int statusCode, String statusMsg )
+ {
+ fWaitMs = waitMs;
+ fStatusCode = statusCode;
+ fStatusMsg = statusMsg;
+ fMsgs = null;
+ }
+
+ public LinkedList<String> run () throws IOException
+ {
+ try
+ {
+ Thread.sleep ( fWaitMs );
+ if ( fStatusCode >= 200 && fStatusCode <= 299 )
+ {
+ return fMsgs;
+ }
+ throw new IOException ( "" + fStatusCode + " " + fStatusMsg );
+ }
+ catch ( InterruptedException e )
+ {
+ throw new IOException ( e );
+ }
+ }
+
+ private final long fWaitMs;
+ private final int fStatusCode;
+ private final String fStatusMsg;
+ private final LinkedList<String> fMsgs;
+ }
+
+ public MRConsumerMock ()
+ {
+ fReplies = new LinkedList<Entry> ();
+ }
+
+ @Override
+ public void close ()
+ {
+ }
+
+ @Override
+ public void setApiCredentials ( String apiKey, String apiSecret )
+ {
+ }
+
+ @Override
+ public void clearApiCredentials ()
+ {
+ }
+
+ public synchronized void add ( Entry e )
+ {
+ fReplies.add ( e );
+ }
+
+ public void addImmediateMsg ( String msg )
+ {
+ addDelayedMsg ( 0, msg );
+ }
+
+ public void addDelayedMsg ( long delay, String msg )
+ {
+ final LinkedList<String> list = new LinkedList<String> ();
+ list.add ( msg );
+ add ( new Entry ( delay, 200, list ) );
+ }
+
+ public void addImmediateMsgGroup ( List<String> msgs )
+ {
+ addDelayedMsgGroup ( 0, msgs );
+ }
+
+ public void addDelayedMsgGroup ( long delay, List<String> msgs )
+ {
+ final LinkedList<String> list = new LinkedList<String> ( msgs );
+ add ( new Entry ( delay, 200, list ) );
+ }
+
+ public void addImmediateError ( int statusCode, String statusText )
+ {
+ add ( new Entry ( 0, statusCode, statusText ) );
+ }
+
+ @Override
+ public Iterable<String> fetch () throws IOException
+ {
+ return fetch ( -1, -1 );
+ }
+
+ @Override
+ public Iterable<String> fetch ( int timeoutMs, int limit ) throws IOException
+ {
+ return fReplies.size () > 0 ? fReplies.removeFirst ().run() : new LinkedList<String>();
+ }
+
+ @Override
+ public void logTo ( Logger log )
+ {
+ }
+
+ private final LinkedList<Entry> fReplies;
+
+ @Override
+ public MRConsumerResponse fetchWithReturnConsumerResponse() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public MRConsumerResponse fetchWithReturnConsumerResponse(int timeoutMs,
+ int limit) {
+ // TODO Auto-generated method stub
+ return null;
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/tools/ApiKeyCommand.java b/src/main/java/com/att/nsa/mr/tools/ApiKeyCommand.java
new file mode 100644
index 0000000..0faefe7
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/tools/ApiKeyCommand.java
@@ -0,0 +1,134 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.tools;
+
+import java.io.IOException;
+import java.io.PrintStream;
+
+import com.att.nsa.apiClient.credentials.ApiCredential;
+import com.att.nsa.apiClient.http.HttpException;
+import com.att.nsa.apiClient.http.HttpObjectNotFoundException;
+import com.att.nsa.cmdtool.Command;
+import com.att.nsa.cmdtool.CommandNotReadyException;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRIdentityManager;
+import com.att.nsa.mr.client.MRClient.MRApiException;
+import com.att.nsa.mr.client.MRIdentityManager.ApiKey;
+
+public class ApiKeyCommand implements Command<MRCommandContext>
+{
+
+ @Override
+ public String[] getMatches ()
+ {
+ return new String[]{
+ "key (create|update) (\\S*) (\\S*)",
+ "key (list) (\\S*)",
+ "key (revoke)",
+ };
+ }
+
+ @Override
+ public void checkReady ( MRCommandContext context ) throws CommandNotReadyException
+ {
+ if ( !context.checkClusterReady () )
+ {
+ throw new CommandNotReadyException ( "Use 'cluster' to specify a cluster to use." );
+ }
+ }
+
+ @Override
+ public void execute ( String[] parts, MRCommandContext context, PrintStream out ) throws CommandNotReadyException
+ {
+ final MRIdentityManager tm = MRClientFactory.createIdentityManager ( context.getCluster(), context.getApiKey(), context.getApiPwd() );
+ context.applyTracer ( tm );
+
+ try
+ {
+ if ( parts[0].equals ( "list" ) )
+ {
+ final ApiKey key = tm.getApiKey ( parts[1] );
+ if ( key != null )
+ {
+ out.println ( "email: " + key.getEmail () );
+ out.println ( "description: " + key.getDescription () );
+ }
+ else
+ {
+ out.println ( "No key returned" );
+ }
+ }
+ else if ( parts[0].equals ( "create" ) )
+ {
+ final ApiCredential ac = tm.createApiKey ( parts[1], parts[2] );
+ if ( ac != null )
+ {
+ out.println ( " key: " + ac.getApiKey () );
+ out.println ( "secret: " + ac.getApiSecret () );
+ }
+ else
+ {
+ out.println ( "No credential returned?" );
+ }
+ }
+ else if ( parts[0].equals ( "update" ) )
+ {
+ tm.updateCurrentApiKey ( parts[1], parts[2] );
+ out.println ( "Updated" );
+ }
+ else if ( parts[0].equals ( "revoke" ) )
+ {
+ tm.deleteCurrentApiKey ();
+ out.println ( "Updated" );
+ }
+ }
+ catch ( HttpObjectNotFoundException e )
+ {
+ out.println ( "Object not found: " + e.getMessage () );
+ }
+ catch ( HttpException e )
+ {
+ out.println ( "HTTP exception: " + e.getMessage () );
+ }
+ catch ( MRApiException e )
+ {
+ out.println ( "API exception: " + e.getMessage () );
+ }
+ catch ( IOException e )
+ {
+ out.println ( "IO exception: " + e.getMessage () );
+ }
+ finally
+ {
+ tm.close ();
+ }
+ }
+
+ @Override
+ public void displayHelp ( PrintStream out )
+ {
+ out.println ( "key create <email> <description>" );
+ out.println ( "key update <email> <description>" );
+ out.println ( "key list <key>" );
+ out.println ( "key revoke" );
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/tools/AuthCommand.java b/src/main/java/com/att/nsa/mr/tools/AuthCommand.java
new file mode 100644
index 0000000..72f226b
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/tools/AuthCommand.java
@@ -0,0 +1,69 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.tools;
+
+import java.io.PrintStream;
+
+import com.att.nsa.cmdtool.Command;
+import com.att.nsa.cmdtool.CommandNotReadyException;
+
+public class AuthCommand implements Command<MRCommandContext>
+{
+ @Override
+ public void checkReady ( MRCommandContext context ) throws CommandNotReadyException
+ {
+ }
+
+ @Override
+ public void execute ( String[] parts, MRCommandContext context, PrintStream out ) throws CommandNotReadyException
+ {
+ if ( parts.length > 0 )
+ {
+ context.setAuth ( parts[0], parts[1] );
+ out.println ( "Now authenticating with " + parts[0] );
+ }
+ else
+ {
+ context.clearAuth ();
+ out.println ( "No longer authenticating." );
+ }
+ }
+
+ @Override
+ public void displayHelp ( PrintStream out )
+ {
+ out.println ( "auth <apiKey> <apiSecret>" );
+ out.println ( "\tuse these credentials on subsequent transactions" );
+ out.println ( "noauth" );
+ out.println ( "\tdo not use credentials on subsequent transactions" );
+ }
+
+ @Override
+ public String[] getMatches ()
+ {
+ return new String[]
+ {
+ "auth (\\S*) (\\S*)",
+ "noauth"
+ };
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/tools/ClusterCommand.java b/src/main/java/com/att/nsa/mr/tools/ClusterCommand.java
new file mode 100644
index 0000000..5ecaf7c
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/tools/ClusterCommand.java
@@ -0,0 +1,80 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.tools;
+
+import java.io.PrintStream;
+
+import com.att.nsa.cmdtool.Command;
+import com.att.nsa.cmdtool.CommandNotReadyException;
+import com.att.nsa.mr.client.impl.MRConsumerImpl;
+
+public class ClusterCommand implements Command<MRCommandContext>
+{
+
+ @Override
+ public String[] getMatches ()
+ {
+ return new String[]{
+ "cluster",
+ "cluster (\\S*)?",
+ };
+ }
+
+ @Override
+ public void checkReady ( MRCommandContext context ) throws CommandNotReadyException
+ {
+ }
+
+ @Override
+ public void execute ( String[] parts, MRCommandContext context, PrintStream out ) throws CommandNotReadyException
+ {
+ if ( parts.length == 0 )
+ {
+ for ( String host : context.getCluster () )
+ {
+ out.println ( host );
+ }
+ }
+ else
+ {
+ context.clearCluster ();
+ for ( String part : parts )
+ {
+ String[] hosts = part.trim().split ( "\\s+" );
+ for ( String host : hosts )
+ {
+ for ( String splitHost : MRConsumerImpl.stringToList(host) )
+ {
+ context.addClusterHost ( splitHost );
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void displayHelp ( PrintStream out )
+ {
+ out.println ( "cluster host1 host2 ..." );
+ }
+
+}
diff --git a/src/main/java/com/att/nsa/mr/tools/MRCommandContext.java b/src/main/java/com/att/nsa/mr/tools/MRCommandContext.java
new file mode 100644
index 0000000..3430d9c
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/tools/MRCommandContext.java
@@ -0,0 +1,100 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.tools;
+
+import java.util.Collection;
+import java.util.LinkedList;
+
+import com.att.nsa.apiClient.http.HttpClient;
+import com.att.nsa.apiClient.http.HttpTracer;
+import com.att.nsa.cmdtool.CommandContext;
+import com.att.nsa.mr.client.MRClient;
+
+public class MRCommandContext implements CommandContext
+{
+ public MRCommandContext ()
+ {
+ fApiKey = null;
+ fApiPwd = null;
+
+ fCluster = new LinkedList<String> ();
+ fCluster.add ( "localhost" );
+ }
+
+ @Override
+ public void requestShutdown ()
+ {
+ fShutdown = true;
+ }
+
+ @Override
+ public boolean shouldContinue ()
+ {
+ return !fShutdown;
+ }
+
+ public void setAuth ( String key, String pwd ) { fApiKey = key; fApiPwd = pwd; }
+ public void clearAuth () { setAuth(null,null); }
+
+ public boolean checkClusterReady ()
+ {
+ return ( fCluster.size () != 0 );
+ }
+
+ public Collection<String> getCluster ()
+ {
+ return new LinkedList<String> ( fCluster );
+ }
+
+ public void clearCluster ()
+ {
+ fCluster.clear ();
+ }
+
+ public void addClusterHost ( String host )
+ {
+ fCluster.add ( host );
+ }
+
+ public String getApiKey () { return fApiKey; }
+ public String getApiPwd () { return fApiPwd; }
+
+ public void useTracer ( HttpTracer t )
+ {
+ fTracer = t;
+ }
+ public void noTracer () { fTracer = null; }
+
+ public void applyTracer ( MRClient cc )
+ {
+ if ( cc instanceof HttpClient && fTracer != null )
+ {
+ ((HttpClient)cc).installTracer ( fTracer );
+ }
+ }
+
+ private boolean fShutdown;
+ private String fApiKey;
+ private String fApiPwd;
+ private final LinkedList<String> fCluster;
+ private HttpTracer fTracer = null;
+}
diff --git a/src/main/java/com/att/nsa/mr/tools/MRTool.java b/src/main/java/com/att/nsa/mr/tools/MRTool.java
new file mode 100644
index 0000000..7f1effd
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/tools/MRTool.java
@@ -0,0 +1,49 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.tools;
+
+import java.io.IOException;
+
+import com.att.nsa.cmdtool.CommandLineTool;
+import com.att.nsa.mr.client.impl.MRClientVersionInfo;
+
+public class MRTool extends CommandLineTool<MRCommandContext>
+{
+ protected MRTool ()
+ {
+ super ( "MR Tool (" + MRClientVersionInfo.getVersion () + ")", "MR> " );
+
+ registerCommand ( new ApiKeyCommand () );
+ registerCommand ( new AuthCommand () );
+ registerCommand ( new ClusterCommand () );
+ registerCommand ( new MessageCommand () );
+ registerCommand ( new TopicCommand () );
+ registerCommand ( new TraceCommand () );
+ }
+
+ public static void main ( String[] args ) throws IOException
+ {
+ final MRTool ct = new MRTool ();
+ final MRCommandContext ccc = new MRCommandContext ();
+ ct.runFromMain ( args, ccc );
+ }
+}
diff --git a/src/main/java/com/att/nsa/mr/tools/MessageCommand.java b/src/main/java/com/att/nsa/mr/tools/MessageCommand.java
new file mode 100644
index 0000000..2c646bf
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/tools/MessageCommand.java
@@ -0,0 +1,128 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.tools;
+
+import java.io.IOException;
+import java.io.PrintStream;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import com.att.nsa.cmdtool.Command;
+import com.att.nsa.cmdtool.CommandNotReadyException;
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRConsumer;
+import com.att.nsa.mr.client.MRClientBuilders.PublisherBuilder;
+import com.att.nsa.mr.client.MRPublisher.message;
+
+public class MessageCommand implements Command<MRCommandContext>
+{
+
+ @Override
+ public String[] getMatches ()
+ {
+ return new String[]{
+ "(post) (\\S*) (\\S*) (.*)",
+ "(read) (\\S*) (\\S*) (\\S*)",
+ };
+ }
+
+ @Override
+ public void checkReady ( MRCommandContext context ) throws CommandNotReadyException
+ {
+ if ( !context.checkClusterReady () )
+ {
+ throw new CommandNotReadyException ( "Use 'cluster' to specify a cluster to use." );
+ }
+ }
+
+ @Override
+ public void execute ( String[] parts, MRCommandContext context, PrintStream out ) throws CommandNotReadyException
+ {
+ if ( parts[0].equalsIgnoreCase ( "read" ))
+ {
+ final MRConsumer cc = MRClientFactory.createConsumer ( context.getCluster (), parts[1], parts[2], parts[3],
+ -1, -1, null, context.getApiKey(), context.getApiPwd() );
+ context.applyTracer ( cc );
+ try
+ {
+ for ( String msg : cc.fetch () )
+ {
+ out.println ( msg );
+ }
+ }
+ catch ( Exception e )
+ {
+ out.println ( "Problem fetching messages: " + e.getMessage() );
+ }
+ finally
+ {
+ cc.close ();
+ }
+ }
+ else
+ {
+ final MRBatchingPublisher pub = new PublisherBuilder ().
+ usingHosts ( context.getCluster () ).
+ onTopic ( parts[1] ).
+ authenticatedBy ( context.getApiKey(), context.getApiPwd() ).
+ build ()
+ ;
+ try
+ {
+ pub.send ( parts[2], parts[3] );
+ }
+ catch ( IOException e )
+ {
+ out.println ( "Problem sending message: " + e.getMessage() );
+ }
+ finally
+ {
+ List<message> left = null;
+ try
+ {
+ left = pub.close ( 500, TimeUnit.MILLISECONDS );
+ }
+ catch ( IOException e )
+ {
+ out.println ( "Problem sending message: " + e.getMessage() );
+ }
+ catch ( InterruptedException e )
+ {
+ out.println ( "Problem sending message: " + e.getMessage() );
+ }
+ if ( left != null && left.size () > 0 )
+ {
+ out.println ( left.size() + " messages not sent." );
+ }
+ }
+ }
+ }
+
+ @Override
+ public void displayHelp ( PrintStream out )
+ {
+ out.println ( "post <topicName> <partition> <message>" );
+ out.println ( "read <topicName> <consumerGroup> <consumerId>" );
+ }
+
+}
diff --git a/src/main/java/com/att/nsa/mr/tools/TopicCommand.java b/src/main/java/com/att/nsa/mr/tools/TopicCommand.java
new file mode 100644
index 0000000..8010a58
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/tools/TopicCommand.java
@@ -0,0 +1,209 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.tools;
+
+import java.io.IOException;
+import java.io.PrintStream;
+import java.util.Set;
+
+import com.att.nsa.apiClient.http.HttpException;
+import com.att.nsa.apiClient.http.HttpObjectNotFoundException;
+import com.att.nsa.cmdtool.Command;
+import com.att.nsa.cmdtool.CommandNotReadyException;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRTopicManager;
+import com.att.nsa.mr.client.MRTopicManager.TopicInfo;
+
+public class TopicCommand implements Command<MRCommandContext>
+{
+
+ @Override
+ public String[] getMatches ()
+ {
+ return new String[]{
+ "topic (list)",
+ "topic (list) (\\S*)",
+ "topic (create) (\\S*) (\\S*) (\\S*)",
+ "topic (grant|revoke) (read|write) (\\S*) (\\S*)",
+ };
+ }
+
+ @Override
+ public void checkReady ( MRCommandContext context ) throws CommandNotReadyException
+ {
+ if ( !context.checkClusterReady () )
+ {
+ throw new CommandNotReadyException ( "Use 'cluster' to specify a cluster to use." );
+ }
+ }
+
+ @Override
+ public void execute ( String[] parts, MRCommandContext context, PrintStream out ) throws CommandNotReadyException
+ {
+ final MRTopicManager tm = MRClientFactory.createTopicManager ( context.getCluster(), context.getApiKey(), context.getApiPwd() );
+ context.applyTracer ( tm );
+
+ try
+ {
+ if ( parts[0].equals ( "list" ) )
+ {
+ try
+ {
+ if ( parts.length == 1 )
+ {
+ for ( String topic : tm.getTopics () )
+ {
+ out.println ( topic );
+ }
+ }
+ else
+ {
+ final TopicInfo ti = tm.getTopicMetadata ( parts[1] );
+
+ final String owner = ti.getOwner ();
+ out.println ( " owner: " + ( owner == null ? "<none>" : owner ) );
+
+ final String desc = ti.getDescription ();
+ out.println ( "description: " + ( desc == null ? "<none>" : desc ) );
+
+ final Set<String> prods = ti.getAllowedProducers ();
+ if ( prods != null )
+ {
+ out.println ( " write ACL: " );
+ for ( String key : prods )
+ {
+ out.println ( "\t" + key );
+ }
+ }
+ else
+ {
+ out.println ( " write ACL: <not active>" );
+ }
+
+ final Set<String> cons = ti.getAllowedConsumers ();
+ if ( cons != null )
+ {
+ out.println ( " read ACL: " );
+ for ( String key : cons )
+ {
+ out.println ( "\t" + key );
+ }
+ }
+ else
+ {
+ out.println ( " read ACL: <not active>" );
+ }
+ }
+ }
+ catch ( IOException x )
+ {
+ out.println ( "Problem with request: " + x.getMessage () );
+ }
+ catch ( HttpObjectNotFoundException e )
+ {
+ out.println ( "Not found: " + e.getMessage () );
+ }
+ }
+ else if ( parts[0].equals ( "create" ) )
+ {
+ try
+ {
+ final int partitions = Integer.parseInt ( parts[2] );
+ final int replicas = Integer.parseInt ( parts[3] );
+
+ tm.createTopic ( parts[1], "", partitions, replicas );
+ }
+ catch ( HttpException e )
+ {
+ out.println ( "Problem with request: " + e.getMessage () );
+ }
+ catch ( IOException e )
+ {
+ out.println ( "Problem with request: " + e.getMessage () );
+ }
+ catch ( NumberFormatException e )
+ {
+ out.println ( "Problem with request: " + e.getMessage () );
+ }
+ }
+ else if ( parts[0].equals ( "grant" ) )
+ {
+ try
+ {
+ if ( parts[1].equals ( "write" ) )
+ {
+ tm.allowProducer ( parts[2], parts[3] );
+ }
+ else if ( parts[1].equals ( "read" ) )
+ {
+ tm.allowConsumer ( parts[2], parts[3] );
+ }
+ }
+ catch ( HttpException e )
+ {
+ out.println ( "Problem with request: " + e.getMessage () );
+ }
+ catch ( IOException e )
+ {
+ out.println ( "Problem with request: " + e.getMessage () );
+ }
+ }
+ else if ( parts[0].equals ( "revoke" ) )
+ {
+ try
+ {
+ if ( parts[1].equals ( "write" ) )
+ {
+ tm.revokeProducer ( parts[2], parts[3] );
+ }
+ else if ( parts[1].equals ( "read" ) )
+ {
+ tm.revokeConsumer ( parts[2], parts[3] );
+ }
+ }
+ catch ( HttpException e )
+ {
+ out.println ( "Problem with request: " + e.getMessage () );
+ }
+ catch ( IOException e )
+ {
+ out.println ( "Problem with request: " + e.getMessage () );
+ }
+ }
+ }
+ finally
+ {
+ tm.close ();
+ }
+ }
+
+ @Override
+ public void displayHelp ( PrintStream out )
+ {
+ out.println ( "topic list" );
+ out.println ( "topic list <topicName>" );
+ out.println ( "topic create <topicName> <partitions> <replicas>" );
+ out.println ( "topic grant write|read <topicName> <apiKey>" );
+ out.println ( "topic revoke write|read <topicName> <apiKey>" );
+ }
+
+}
diff --git a/src/main/java/com/att/nsa/mr/tools/TraceCommand.java b/src/main/java/com/att/nsa/mr/tools/TraceCommand.java
new file mode 100644
index 0000000..0489172
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/tools/TraceCommand.java
@@ -0,0 +1,118 @@
+/*******************************************************************************
+ * ============LICENSE_START=======================================================
+ * org.onap.dmaap
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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 com.att.nsa.mr.tools;
+
+import java.io.PrintStream;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+
+import com.att.nsa.apiClient.http.HttpTracer;
+import com.att.nsa.cmdtool.Command;
+import com.att.nsa.cmdtool.CommandNotReadyException;
+
+public class TraceCommand implements Command<MRCommandContext>
+{
+ @Override
+ public void checkReady ( MRCommandContext context ) throws CommandNotReadyException
+ {
+ }
+
+ @Override
+ public void execute ( String[] parts, MRCommandContext context, final PrintStream out ) throws CommandNotReadyException
+ {
+ if ( parts[0].equalsIgnoreCase ( "on" ))
+ {
+ context.useTracer ( new HttpTracer ()
+ {
+ @Override
+ public void outbound ( URI uri, Map<String, List<String>> headers, String method, byte[] entity )
+ {
+ out.println ( kLineBreak );
+ out.println ( ">>> " + method + " " + uri.toString() );
+ for ( Map.Entry<String,List<String>> e : headers.entrySet () )
+ {
+ final StringBuffer vals = new StringBuffer ();
+ for ( String val : e.getValue () )
+ {
+ if ( vals.length () > 0 ) vals.append ( ", " );
+ vals.append ( val );
+ }
+ out.println ( ">>> " + e.getKey () + ": " + vals.toString() );
+ }
+ if ( entity != null )
+ {
+ out.println ();
+ out.println ( new String ( entity ) );
+ }
+ out.println ( kLineBreak );
+ }
+
+ @Override
+ public void inbound ( Map<String, List<String>> headers, int statusCode, String responseLine, byte[] entity )
+ {
+ out.println ( kLineBreak );
+ out.println ( "<<< " + responseLine );
+ for ( Map.Entry<String,List<String>> e : headers.entrySet () )
+ {
+ final StringBuffer vals = new StringBuffer ();
+ for ( String val : e.getValue () )
+ {
+ if ( vals.length () > 0 ) vals.append ( ", " );
+ vals.append ( val );
+ }
+ out.println ( "<<< " + e.getKey () + ": " + vals.toString() );
+ }
+ if ( entity != null )
+ {
+ out.println ();
+ out.println ( new String ( entity ) );
+ }
+ out.println ( kLineBreak );
+ }
+ } );
+ }
+ else
+ {
+ context.noTracer ();
+ }
+ }
+
+ @Override
+ public void displayHelp ( PrintStream out )
+ {
+ out.println ( "trace on|off" );
+ out.println ( "\tWhen trace is on, HTTP interaction is printed to the console." );
+ }
+
+ @Override
+ public String[] getMatches ()
+ {
+ return new String[]
+ {
+ "trace (on)",
+ "trace (off)"
+ };
+ }
+
+ private static final String kLineBreak = "======================================================================";
+}