From 7d45c179879363222fcf49b30f75837f66d7f423 Mon Sep 17 00:00:00 2001 From: Varun Gudisena Date: Thu, 31 Aug 2017 10:44:28 -0500 Subject: 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 --- .../java/com/att/nsa/mr/client/HostSelector.java | 191 +++++ .../com/att/nsa/mr/client/MRBatchingPublisher.java | 55 ++ src/main/java/com/att/nsa/mr/client/MRClient.java | 66 ++ .../com/att/nsa/mr/client/MRClientBuilders.java | 383 +++++++++ .../com/att/nsa/mr/client/MRClientFactory.java | 558 ++++++++++++ .../java/com/att/nsa/mr/client/MRConsumer.java | 54 ++ .../com/att/nsa/mr/client/MRIdentityManager.java | 100 +++ .../java/com/att/nsa/mr/client/MRPublisher.java | 93 ++ .../java/com/att/nsa/mr/client/MRTopicManager.java | 183 ++++ .../java/com/att/nsa/mr/client/impl/Clock.java | 63 ++ .../com/att/nsa/mr/client/impl/MRBaseClient.java | 393 +++++++++ .../att/nsa/mr/client/impl/MRBatchPublisher.java | 485 +++++++++++ .../nsa/mr/client/impl/MRClientVersionInfo.java | 54 ++ .../com/att/nsa/mr/client/impl/MRConstants.java | 180 ++++ .../com/att/nsa/mr/client/impl/MRConsumerImpl.java | 709 ++++++++++++++++ .../java/com/att/nsa/mr/client/impl/MRFormat.java | 49 ++ .../com/att/nsa/mr/client/impl/MRMetaClient.java | 260 ++++++ .../mr/client/impl/MRSimplerBatchPublisher.java | 939 +++++++++++++++++++++ .../nsa/mr/client/response/MRConsumerResponse.java | 60 ++ .../mr/client/response/MRPublisherResponse.java | 66 ++ 20 files changed, 4941 insertions(+) create mode 100644 src/main/java/com/att/nsa/mr/client/HostSelector.java create mode 100644 src/main/java/com/att/nsa/mr/client/MRBatchingPublisher.java create mode 100644 src/main/java/com/att/nsa/mr/client/MRClient.java create mode 100644 src/main/java/com/att/nsa/mr/client/MRClientBuilders.java create mode 100644 src/main/java/com/att/nsa/mr/client/MRClientFactory.java create mode 100644 src/main/java/com/att/nsa/mr/client/MRConsumer.java create mode 100644 src/main/java/com/att/nsa/mr/client/MRIdentityManager.java create mode 100644 src/main/java/com/att/nsa/mr/client/MRPublisher.java create mode 100644 src/main/java/com/att/nsa/mr/client/MRTopicManager.java create mode 100644 src/main/java/com/att/nsa/mr/client/impl/Clock.java create mode 100644 src/main/java/com/att/nsa/mr/client/impl/MRBaseClient.java create mode 100644 src/main/java/com/att/nsa/mr/client/impl/MRBatchPublisher.java create mode 100644 src/main/java/com/att/nsa/mr/client/impl/MRClientVersionInfo.java create mode 100644 src/main/java/com/att/nsa/mr/client/impl/MRConstants.java create mode 100644 src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java create mode 100644 src/main/java/com/att/nsa/mr/client/impl/MRFormat.java create mode 100644 src/main/java/com/att/nsa/mr/client/impl/MRMetaClient.java create mode 100644 src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java create mode 100644 src/main/java/com/att/nsa/mr/client/response/MRConsumerResponse.java create mode 100644 src/main/java/com/att/nsa/mr/client/response/MRPublisherResponse.java (limited to 'src/main/java/com/att/nsa/mr/client') 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 fBaseHosts; + private final DelayQueue 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 baseHosts) + { + this(baseHosts, null); + } + + public HostSelector(Collection 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 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 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 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 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 hosts = new TreeSet (); + 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 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 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 + { + /** + * Construct an identity manager builder. + */ + public IdentityManagerBuilder () {} + + @Override + protected MRIdentityManager constructClient ( Collection 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 + { + /** + * Construct an topic manager builder. + */ + public TopicManagerBuilder () {} + + @Override + protected MRTopicManager constructClient ( Collection 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 + { + /** + * 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 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 usingHosts ( Collection hostSet ) { fHosts = hostSet; return this; } + + /** + * Set the API key and secret for this client. + * @param apiKey + * @param apiSecret + * @return this builder + */ + public AbstractAuthenticatedManagerBuilder 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 hosts ); + + private Collection 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.
+ *
+ * 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.)
+ *
+ * Publishers + * + * @author author + */ +public class MRClientFactory +{ + public static MultivaluedMap HTTPHeadersMap; + public static Map 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 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 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 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 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 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 hosts = new TreeSet (); + 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 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 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 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 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 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 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 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 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 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 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 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 hosts ) throws MalformedURLException + { + super ( ConnectionType.HTTP, hosts, MRConstants.kStdMRServicePort ); + } + + protected MRBaseClient ( Collection hosts, int stdSvcPort ) throws MalformedURLException { + super ( ConnectionType.HTTP,hosts, stdSvcPort); + + fLog = LoggerFactory.getLogger ( this.getClass().getName () ); + } + + protected MRBaseClient ( Collection 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 jsonArrayToSet ( JSONArray a ) + { + if ( a == null ) return null; + + final TreeSet set = new TreeSet (); + for ( int i=0; i 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("")){ + + beginIndex = responseString.indexOf("body>")+5; + endIndex = responseString.indexOf("")){ + beginIndex = responseString.indexOf("title>")+6; + endIndex = responseString.indexOf(" 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 list = new LinkedList (); + 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 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 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 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 result = new LinkedList (); + 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 baseUrls, String topic, int maxBatch, long maxAgeMs, boolean compress ) throws MalformedURLException + { + super ( baseUrls ); + + fNextBatch = new LinkedList (); + 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 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 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 (); + } + 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 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 fNextBatch; + private LinkedList 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 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 createHostsList(Collection hosts) + { + final ArrayList convertedHosts = new ArrayList (); + 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 stringToList ( String str ) + { + final LinkedList set = new LinkedList (); + 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 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 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 fetch () throws IOException,Exception + { + // fetch with the timeout and limit set in constructor + return fetch ( fTimeoutMs, fLimit ); + } + + @Override + public Iterable fetch ( int timeoutMs, int limit ) throws IOException,Exception + { + final LinkedList msgs = new LinkedList (); + +// 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; i0 ) + if ( a != null) + { + for ( int i=0; i0) + if ( a != null) + { + for ( int i=0; i 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(); + 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 msgs = new LinkedList(); + 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 baseUrls ) throws MalformedURLException + { + super ( baseUrls ); + } + + @Override + public Set getTopics () throws IOException + { + final TreeSet set = new TreeSet (); + try + { + final JSONObject topicSet = get ( "/topics" ); + final JSONArray a = topicSet.getJSONArray ( "topics" ); + for ( int i=0; i getAllowedProducers () + { + final JSONObject acl = topicData.optJSONObject ( "writerAcl" ); + if ( acl != null && acl.optBoolean ( "enabled", true ) ) + { + return jsonArrayToSet ( acl.optJSONArray ( "users" ) ); + } + return null; + } + + @Override + public Set 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 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 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 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 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 list = new LinkedList (); + list.add ( msg ); + return send ( list ); + } + + + + @Override + public synchronized int send ( Collection 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 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 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 result = new LinkedList (); + 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 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 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 headers=new HashMap(); + public static MultivaluedMap 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(); + 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 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 (); + fDontSendUntilMs = 0; + fExec = new ScheduledThreadPoolExecutor ( 1 ); + pubResponse = new MRPublisherResponse(); + + } + + private MRSimplerBatchPublisher ( Collection 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 (); + 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 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 getActualMessages() { + return actualMessages; + } + + public void setActualMessages(Iterable 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; + } + +} -- cgit 1.2.3-korg