* Copyright © 2017 AT&T Intellectual Property. All rights reserved.
* ================================================================================
* Modifications Copyright © 2018 IBM.
+ * Modifications Copyright © 2021 Orange.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* ============LICENSE_END=========================================================
*
* ECOMP is a trademark and service mark of AT&T Intellectual Property.
- *
+ *
*******************************************************************************/
+
package org.onap.dmaap.mr.client;
import java.net.MalformedURLException;
import java.util.Collection;
+import java.util.Collections;
import java.util.TreeSet;
import java.util.UUID;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import org.onap.dmaap.mr.client.impl.MRConsumerImpl;
import org.onap.dmaap.mr.client.impl.MRMetaClient;
import org.onap.dmaap.mr.client.impl.MRSimplerBatchPublisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
- * A collection of builders for various types of MR API clients
- *
+ * A collection of builders for various types of MR API clients.
+ *
* @author author
*/
-public class MRClientBuilders
-{
- private final static String ILLEGAL_ARGUMENT_MESSAGE = "You must provide at least one host and a topic name.";
+public class MRClientBuilders {
+ private static final Logger logger = LoggerFactory.getLogger(MRClientBuilders.class);
+
+ private static final String ILLEGAL_ARGUMENT_MESSAGE = "You must provide at least one host and a topic name.";
/**
* Instantiates MRClientBuilders.
private MRClientBuilders() {
// prevent instantiation
}
-
+
/**
- * A builder for a topic Consumer
+ * A builder for a topic Consumer.
+ *
* @author author
*/
- public static class ConsumerBuilder
- {
+ public static class ConsumerBuilder {
/**
- * Set the host list
+ * 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) );
+ public ConsumerBuilder usingHosts(String hostList) {
+ return usingHosts(MRConsumerImpl.stringToList(hostList));
}
/**
- * Set the host list
+ * Set the host list.
+ *
* @param hostSet a set of hosts to use to connect to MR
* @return this builder
*/
- public ConsumerBuilder usingHosts ( Collection<String> hostSet ) {
- fHosts = hostSet; return this;
+ public ConsumerBuilder usingHosts(Collection<String> hostSet) {
+ fHosts = hostSet;
+ return this;
}
/**
- * Set the topic
+ * Set the topic.
+ *
* @param topic the name of the topic to consume
* @return this builder
*/
- public ConsumerBuilder onTopic ( String topic ) {
- fTopic=topic;
- return this;
+ public ConsumerBuilder onTopic(String topic) {
+ fTopic = topic;
+ return this;
}
/**
- * Set the consumer's group and ID
+ * 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
+ * @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;
+ 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;
+ 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.
+ * 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;
- };
+ public ConsumerBuilder waitAtServer(int timeoutMs) {
+ fTimeoutMs = timeoutMs;
+ return this;
+ }
/**
- * Set the maximum number of messages to receive per transaction
+ * 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;
- };
+ public ConsumerBuilder receivingAtMost(int limit) {
+ fLimit = limit;
+ return this;
+ }
/**
- * Set a filter to use on the server
+ * 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;
+ public ConsumerBuilder withServerSideFilter(String filter) {
+ fFilter = filter;
+ return this;
}
/**
- * Build the consumer
+ * Build the consumer.
+ *
* @return a consumer
*/
- public MRConsumer build ()
- {
- if ( fHosts == null || fHosts.isEmpty() || fTopic == null )
- {
- throw new IllegalArgumentException ( ILLEGAL_ARGUMENT_MESSAGE );
+ public MRConsumer build() {
+ if (fHosts == null || fHosts.isEmpty() || fTopic == null) {
+ throw new IllegalArgumentException(ILLEGAL_ARGUMENT_MESSAGE);
}
- if ( fGroup == null )
- {
- fGroup = UUID.randomUUID ().toString ();
+ if (fGroup == null) {
+ fGroup = UUID.randomUUID().toString();
fId = "0";
- log.info ( "Creating non-restartable client with group " + fGroup + " and ID " + fId + "." );
+ logger.info("Creating non-restartable client with group {} and ID {}.", fGroup, fId);
}
- if ( sfConsumerMock != null ) return sfConsumerMock;
+ if (sfConsumerMock != null) {
+ return sfConsumerMock;
+ }
try {
return new MRConsumerImpl.MRConsumerImplBuilder().setHostPart(fHosts)
.setTopic(fTopic).setConsumerGroup(fGroup).setConsumerId(fId)
private String fFilter = null;
}
- /*************************************************************************/
- /*************************************************************************/
- /*************************************************************************/
+ //*************************************************************************/
+ //*************************************************************************/
+ //*************************************************************************/
/**
- * A publisher builder
+ * A publisher builder.
+ *
* @author author
*/
- public static class PublisherBuilder
- {
+ public static class PublisherBuilder {
/**
- * Set the MR/UEB host(s) to use
+ * 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) );
+ public PublisherBuilder usingHosts(String hostlist) {
+ return usingHosts(MRConsumerImpl.stringToList(hostlist));
}
/**
- * Set the MR/UEB host(s) to use
+ * Set the MR/UEB host(s) to use.
+ *
* @param hostSet The host(s) used in the URL to MR. Can be "host:port"
* @return this builder
*/
- public PublisherBuilder usingHosts ( String[] hostSet )
- {
- final TreeSet<String> hosts = new TreeSet<> ();
- for ( String hp : hostSet )
- {
- hosts.add ( hp );
- }
- return usingHosts ( hosts );
+ public PublisherBuilder usingHosts(String[] hostSet) {
+ final TreeSet<String> hosts = new TreeSet<>();
+ Collections.addAll(hosts, hostSet);
+ return usingHosts(hosts);
}
/**
- * Set the MR/UEB host(s) to use
+ * Set the MR/UEB host(s) to use.
+ *
* @param hostlist The host(s) used in the URL to MR. Can be "host:port".
* @return this builder
*/
- public PublisherBuilder usingHosts ( Collection<String> hostlist ) {
- fHosts=hostlist;
- return this;
+ public PublisherBuilder usingHosts(Collection<String> hostlist) {
+ fHosts = hostlist;
+ return this;
}
/**
- * Set the topic to publish on
+ * 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;
+ 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.
+ * @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;
+ public PublisherBuilder limitBatch(int messageCount, int ageInMs) {
+ fMaxBatchSize = messageCount;
+ fMaxBatchAgeMs = ageInMs;
+ return this;
}
/**
- * Compress transactions
+ * Compress transactions.
+ *
* @return this builder
*/
- public PublisherBuilder withCompresion () {
- return enableCompresion(true);
+ public PublisherBuilder withCompresion() {
+ return enableCompresion(true);
}
/**
- * Do not compress transactions
+ * Do not compress transactions.
+ *
* @return this builder
*/
- public PublisherBuilder withoutCompresion () {
- return enableCompresion(false);
+ public PublisherBuilder withoutCompresion() {
+ return enableCompresion(false);
}
/**
- * Set the compression option
+ * Set the compression option.
+ *
* @param compress true to gzip compress transactions
* @return this builder
*/
- public PublisherBuilder enableCompresion ( boolean compress ) {
- fCompress = compress;
- return this;
+ 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;
+ public PublisherBuilder authenticatedBy(String apiKey, String apiSecret) {
+ fApiKey = apiKey;
+ fApiSecret = apiSecret;
+ return this;
}
/**
- * Build the publisher
+ * Build the publisher.
+ *
* @return a batching publisher
*/
- public MRBatchingPublisher build ()
- {
- if ( fHosts == null || fHosts.isEmpty() || fTopic == null )
- {
- throw new IllegalArgumentException ( ILLEGAL_ARGUMENT_MESSAGE );
+ public MRBatchingPublisher build() {
+ if (fHosts == null || fHosts.isEmpty() || fTopic == null) {
+ throw new IllegalArgumentException(ILLEGAL_ARGUMENT_MESSAGE);
}
- 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 );
+ if (sfPublisherMock != null) {
+ return sfPublisherMock;
+ }
+
+ final MRSimplerBatchPublisher pub = new MRSimplerBatchPublisher.Builder()
+ .againstUrls(fHosts)
+ .onTopic(fTopic)
+ .batchTo(fMaxBatchSize, fMaxBatchAgeMs)
+ .compress(fCompress)
+ .build();
+ if (fApiKey != null) {
+ pub.setApiCredentials(fApiKey, fApiSecret);
}
return pub;
}
-
+
private Collection<String> fHosts = null;
private String fTopic = null;
private int fMaxBatchSize = 1;
}
/**
- * A builder for an identity manager
+ * A builder for an identity manager.
+ *
* @author author
*/
- public static class IdentityManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRIdentityManager>
- {
+ public static class IdentityManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRIdentityManager> {
@Override
- protected MRIdentityManager constructClient ( Collection<String> hosts ) { try {
- return new MRMetaClient ( hosts );
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException(e);
- } }
+ protected MRIdentityManager constructClient(Collection<String> hosts) {
+ try {
+ return new MRMetaClient(hosts);
+ } catch (MalformedURLException e) {
+ throw new IllegalArgumentException(e);
+ }
+ }
}
/**
- * A builder for a topic manager
+ * A builder for a topic manager.
+ *
* @author author
*/
- public static class TopicManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRTopicManager>
- {
+ public static class TopicManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRTopicManager> {
@Override
- protected MRTopicManager constructClient ( Collection<String> hosts ) { try {
- return new MRMetaClient ( hosts );
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException(e);
- } }
+ protected MRTopicManager constructClient(Collection<String> hosts) {
+ try {
+ return new MRMetaClient(hosts);
+ } catch (MalformedURLException e) {
+ throw new IllegalArgumentException(e);
+ }
+ }
}
/**
* Inject a consumer. Used to support unit tests.
+ *
* @param cc
*/
- public static void $testInject ( MRConsumer 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 )
- {
+ public static void $testInject(MRBatchingPublisher pub) {
sfPublisherMock = pub;
}
static MRBatchingPublisher sfPublisherMock = null;
/**
- * A builder for an identity manager
+ * A builder for an identity manager.
+ *
* @author author
*/
- public static abstract class AbstractAuthenticatedManagerBuilder<T extends MRClient>
- {
- /**
- * Construct an identity manager builder.
- */
- public AbstractAuthenticatedManagerBuilder () {}
+ public abstract static class AbstractAuthenticatedManagerBuilder<T extends MRClient> {
/**
- * Set the host list
+ * Set the host list.
+ *
* @param hostList a comma-separated list of hosts to use to connect to MR
* @return this builder
*/
- public AbstractAuthenticatedManagerBuilder<T> usingHosts ( String hostList ) {
- return usingHosts ( MRConsumerImpl.stringToList(hostList) );
+ public AbstractAuthenticatedManagerBuilder<T> usingHosts(String hostList) {
+ return usingHosts(MRConsumerImpl.stringToList(hostList));
}
/**
- * Set the host list
+ * Set the host list.
+ *
* @param hostSet a set of hosts to use to connect to MR
* @return this builder
*/
- public AbstractAuthenticatedManagerBuilder<T> usingHosts ( Collection<String> hostSet ) {
- fHosts = hostSet;
- return this;
+ public AbstractAuthenticatedManagerBuilder<T> usingHosts(Collection<String> hostSet) {
+ fHosts = hostSet;
+ return this;
}
/**
* Set the API key and secret for this client.
+ *
* @param apiKey
* @param apiSecret
* @return this builder
*/
- public AbstractAuthenticatedManagerBuilder<T> authenticatedBy ( String apiKey, String apiSecret ) {
- fApiKey = apiKey;
- fApiSecret = apiSecret;
- return this;
+ public AbstractAuthenticatedManagerBuilder<T> authenticatedBy(String apiKey, String apiSecret) {
+ fApiKey = apiKey;
+ fApiSecret = apiSecret;
+ return this;
}
/**
- * Build the consumer
+ * Build the consumer.
+ *
* @return a consumer
*/
- public T build ()
- {
- if ( fHosts.isEmpty() )
- {
- throw new IllegalArgumentException ( ILLEGAL_ARGUMENT_MESSAGE );
+ public T build() {
+ if (fHosts.isEmpty()) {
+ throw new IllegalArgumentException(ILLEGAL_ARGUMENT_MESSAGE);
}
- final T mgr = constructClient ( fHosts );
- mgr.setApiCredentials ( fApiKey, fApiSecret );
+ final T mgr = constructClient(fHosts);
+ mgr.setApiCredentials(fApiKey, fApiSecret);
return mgr;
}
- protected abstract T constructClient ( Collection<String> hosts );
+ protected abstract T constructClient(Collection<String> hosts);
private Collection<String> fHosts = null;
private String fApiKey = null;
private String fApiSecret = null;
}
-
- private static final Logger log = LoggerFactory.getLogger ( MRClientBuilders.class );
+
}
Code Review / dmaap / messagerouter / dmaapclient.git / blobdiff