* ================================================================================
* Copyright © 2017 AT&T Intellectual Property. All rights reserved.
* ================================================================================
+ * Modifications Copyright © 2018 IBM.
+ * ================================================================================
* 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
*/
public class MRClientBuilders
{
- /**
- * A builder for a topic Consumer
- * @author author
- */
- public static class ConsumerBuilder
- {
- /**
- * Construct a consumer builder.
- */
- public ConsumerBuilder () {}
-
- /**
- * Set the host list
- * @param hostList a comma-separated list of hosts to use to connect to MR
- * @return this builder
- */
- public ConsumerBuilder usingHosts ( String hostList ) { return usingHosts ( MRConsumerImpl.stringToList(hostList) ); }
-
- /**
- * Set the host list
- * @param hostSet a set of hosts to use to connect to MR
- * @return this builder
- */
- public ConsumerBuilder usingHosts ( Collection<String> hostSet ) { fHosts = hostSet; return this; }
-
- /**
- * Set the topic
- * @param topic the name of the topic to consume
- * @return this builder
- */
- public ConsumerBuilder onTopic ( String topic ) { fTopic=topic; return this; }
-
- /**
- * Set the consumer's group and ID
- * @param consumerGroup The name of the consumer group this consumer is part of
- * @param consumerId The unique id of this consumer in its group
- * @return this builder
- */
- public ConsumerBuilder knownAs ( String consumerGroup, String consumerId ) { fGroup = consumerGroup; fId = consumerId; return this; }
-
- /**
- * Set the API key and secret for this client.
- * @param apiKey
- * @param apiSecret
- * @return this builder
- */
- public ConsumerBuilder authenticatedBy ( String apiKey, String apiSecret ) { fApiKey = apiKey; fApiSecret = apiSecret; return this; }
-
- /**
- * Set the server side timeout
- * @param timeoutMs The amount of time in milliseconds that the server should keep the connection open while waiting for message traffic.
- * @return this builder
- */
- public ConsumerBuilder waitAtServer ( int timeoutMs ) { fTimeoutMs = timeoutMs; return this; };
-
- /**
- * Set the maximum number of messages to receive per transaction
- * @param limit The maximum number of messages to receive from the server in one transaction.
- * @return this builder
- */
- public ConsumerBuilder receivingAtMost ( int limit ) { fLimit = limit; return this; };
-
- /**
- * Set a filter to use on the server
- * @param filter a Highland Park standard library filter encoded in JSON
- * @return this builder
- */
- public ConsumerBuilder withServerSideFilter ( String filter ) { fFilter = filter; return this; }
-
- /**
- * Build the consumer
- * @return a consumer
- */
- public MRConsumer build ()
- {
- if ( fHosts == null || fHosts.size() == 0 || fTopic == null )
- {
- throw new IllegalArgumentException ( "You must provide at least one host and a topic name." );
- }
-
- if ( fGroup == null )
- {
- fGroup = UUID.randomUUID ().toString ();
- fId = "0";
- log.info ( "Creating non-restartable client with group " + fGroup + " and ID " + fId + "." );
- }
-
- if ( sfConsumerMock != null ) return sfConsumerMock;
- try {
- return new MRConsumerImpl ( fHosts, fTopic, fGroup, fId, fTimeoutMs, fLimit, fFilter, fApiKey, fApiSecret );
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException(e);
- }
- }
-
- private Collection<String> fHosts = null;
- private String fTopic = null;
- private String fGroup = null;
- private String fId = null;
- private String fApiKey = null;
- private String fApiSecret = null;
- private int fTimeoutMs = -1;
- private int fLimit = -1;
- private String fFilter = null;
- }
-
- /*************************************************************************/
- /*************************************************************************/
- /*************************************************************************/
-
- /**
- * A publisher builder
- * @author author
- */
- public static class PublisherBuilder
- {
- public PublisherBuilder () {}
-
- /**
- * Set the MR/UEB host(s) to use
- * @param hostlist The host(s) used in the URL to MR. Can be "host:port", can be multiple comma-separated entries.
- * @return this builder
- */
- public PublisherBuilder usingHosts ( String hostlist ) { return usingHosts ( MRConsumerImpl.stringToList(hostlist) ); }
-
- /**
- * Set the MR/UEB host(s) to use
- * @param hostSet The host(s) used in the URL to MR. Can be "host:port"
- * @return this builder
- */
- public PublisherBuilder usingHosts ( String[] hostSet )
- {
- final TreeSet<String> hosts = new TreeSet<String> ();
- for ( String hp : hostSet )
- {
- hosts.add ( hp );
- }
- return usingHosts ( hosts );
- }
-
- /**
- * Set the MR/UEB host(s) to use
- * @param hostlist The host(s) used in the URL to MR. Can be "host:port".
- * @return this builder
- */
- public PublisherBuilder usingHosts ( Collection<String> hostlist ) { fHosts=hostlist; return this; }
-
- /**
- * Set the topic to publish on
- * @param topic The topic on which to publish messages.
- * @return this builder
- */
- public PublisherBuilder onTopic ( String topic ) { fTopic = topic; return this; }
-
- /**
- * Batch message sends with the given limits.
- * @param messageCount The largest set of messages to batch.
- * @param ageInMs The maximum age of a message waiting in a batch.
- * @return this builder
- */
- public PublisherBuilder limitBatch ( int messageCount, int ageInMs ) { fMaxBatchSize = messageCount; fMaxBatchAgeMs = ageInMs; return this; }
-
- /**
- * Compress transactions
- * @return this builder
- */
- public PublisherBuilder withCompresion () { return enableCompresion(true); }
-
- /**
- * Do not compress transactions
- * @return this builder
- */
- public PublisherBuilder withoutCompresion () { return enableCompresion(false); }
-
- /**
- * Set the compression option
- * @param compress true to gzip compress transactions
- * @return this builder
- */
- public PublisherBuilder enableCompresion ( boolean compress ) { fCompress = compress; return this; }
-
- /**
- * Set the API key and secret for this client.
- * @param apiKey
- * @param apiSecret
- * @return this builder
- */
- public PublisherBuilder authenticatedBy ( String apiKey, String apiSecret ) { fApiKey = apiKey; fApiSecret = apiSecret; return this; }
-
- /**
- * Build the publisher
- * @return a batching publisher
- */
- public MRBatchingPublisher build ()
- {
- if ( fHosts == null || fHosts.isEmpty() || fTopic == null )
- {
- throw new IllegalArgumentException ( "You must provide at least one host and a topic name." );
- }
-
- if ( sfPublisherMock != null ) return sfPublisherMock;
-
- final MRSimplerBatchPublisher pub = new MRSimplerBatchPublisher.Builder ().
- againstUrls ( fHosts ).
- onTopic ( fTopic ).
- batchTo ( fMaxBatchSize, fMaxBatchAgeMs ).
- compress ( fCompress ).
- build ();
- if ( fApiKey != null )
- {
- pub.setApiCredentials ( fApiKey, fApiSecret );
- }
- return pub;
- }
-
- private Collection<String> fHosts = null;
- private String fTopic = null;
- private int fMaxBatchSize = 1;
- private int fMaxBatchAgeMs = 1;
- private boolean fCompress = false;
- private String fApiKey = null;
- private String fApiSecret = null;
- }
-
- /**
- * A builder for an identity manager
- * @author author
- */
- public static class IdentityManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRIdentityManager>
- {
- /**
- * Construct an identity manager builder.
- */
- public IdentityManagerBuilder () {}
-
- @Override
- protected MRIdentityManager constructClient ( Collection<String> hosts ) { try {
- return new MRMetaClient ( hosts );
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException(e);
- } }
- }
-
- /**
- * A builder for a topic manager
- * @author author
- */
- public static class TopicManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRTopicManager>
- {
- /**
- * Construct an topic manager builder.
- */
- public TopicManagerBuilder () {}
-
- @Override
- protected MRTopicManager constructClient ( Collection<String> hosts ) { try {
- return new MRMetaClient ( hosts );
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException(e);
- } }
- }
-
- /**
- * Inject a consumer. Used to support unit tests.
- * @param cc
- */
- public static void $testInject ( MRConsumer cc )
- {
- sfConsumerMock = cc;
- }
-
- /**
- * Inject a publisher. Used to support unit tests.
- * @param pub
- */
- public static void $testInject ( MRBatchingPublisher pub )
- {
- sfPublisherMock = pub;
- }
-
- static MRConsumer sfConsumerMock = null;
- static MRBatchingPublisher sfPublisherMock = null;
-
- /**
- * A builder for an identity manager
- * @author author
- */
- public static abstract class AbstractAuthenticatedManagerBuilder<T extends MRClient>
- {
- /**
- * Construct an identity manager builder.
- */
- public AbstractAuthenticatedManagerBuilder () {}
-
- /**
- * Set the host list
- * @param hostList a comma-separated list of hosts to use to connect to MR
- * @return this builder
- */
- public AbstractAuthenticatedManagerBuilder<T> usingHosts ( String hostList ) { return usingHosts ( MRConsumerImpl.stringToList(hostList) ); }
-
- /**
- * Set the host list
- * @param hostSet a set of hosts to use to connect to MR
- * @return this builder
- */
- public AbstractAuthenticatedManagerBuilder<T> usingHosts ( Collection<String> hostSet ) { fHosts = hostSet; return this; }
-
- /**
- * Set the API key and secret for this client.
- * @param apiKey
- * @param apiSecret
- * @return this builder
- */
- public AbstractAuthenticatedManagerBuilder<T> authenticatedBy ( String apiKey, String apiSecret ) { fApiKey = apiKey; fApiSecret = apiSecret; return this; }
-
- /**
- * Build the consumer
- * @return a consumer
- */
- public T build ()
- {
- if ( fHosts.isEmpty() )
- {
- throw new IllegalArgumentException ( "You must provide at least one host and a topic name." );
- }
-
- final T mgr = constructClient ( fHosts );
- mgr.setApiCredentials ( fApiKey, fApiSecret );
- return mgr;
- }
-
- protected abstract T constructClient ( Collection<String> hosts );
-
- private Collection<String> fHosts = null;
- private String fApiKey = null;
- private String fApiSecret = null;
- }
-
- private static final Logger log = LoggerFactory.getLogger ( MRClientBuilders.class );
+
+ /**
+ * Instantiates MRClientBuilders.
+ */
+ private MRClientBuilders() {
+ // prevent instantiation
+ }
+
+ /**
+ * A builder for a topic Consumer
+ * @author author
+ */
+ public static class ConsumerBuilder
+ {
+ /**
+ * Construct a consumer builder.
+ */
+ public ConsumerBuilder () {}
+
+ /**
+ * Set the host list
+ * @param hostList a comma-separated list of hosts to use to connect to MR
+ * @return this builder
+ */
+ public ConsumerBuilder usingHosts ( String hostList ) {
+ return usingHosts ( MRConsumerImpl.stringToList(hostList) );
+ }
+
+ /**
+ * Set the host list
+ * @param hostSet a set of hosts to use to connect to MR
+ * @return this builder
+ */
+ public ConsumerBuilder usingHosts ( Collection<String> hostSet ) {
+ fHosts = hostSet; return this;
+ }
+
+ /**
+ * Set the topic
+ * @param topic the name of the topic to consume
+ * @return this builder
+ */
+ public ConsumerBuilder onTopic ( String topic ) {
+ fTopic=topic;
+ return this;
+ }
+
+ /**
+ * Set the consumer's group and ID
+ * @param consumerGroup The name of the consumer group this consumer is part of
+ * @param consumerId The unique id of this consumer in its group
+ * @return this builder
+ */
+ public ConsumerBuilder knownAs ( String consumerGroup, String consumerId ) {
+ fGroup = consumerGroup;
+ fId = consumerId;
+ return this;
+ }
+
+ /**
+ * Set the API key and secret for this client.
+ * @param apiKey
+ * @param apiSecret
+ * @return this builder
+ */
+ public ConsumerBuilder authenticatedBy ( String apiKey, String apiSecret ) {
+ fApiKey = apiKey;
+ fApiSecret = apiSecret;
+ return this;
+ }
+
+ /**
+ * Set the server side timeout
+ * @param timeoutMs The amount of time in milliseconds that the server should keep the connection open while waiting for message traffic.
+ * @return this builder
+ */
+ public ConsumerBuilder waitAtServer ( int timeoutMs ) {
+ fTimeoutMs = timeoutMs;
+ return this;
+ };
+
+ /**
+ * Set the maximum number of messages to receive per transaction
+ * @param limit The maximum number of messages to receive from the server in one transaction.
+ * @return this builder
+ */
+ public ConsumerBuilder receivingAtMost ( int limit ) {
+ fLimit = limit;
+ return this;
+ };
+
+ /**
+ * Set a filter to use on the server
+ * @param filter a Highland Park standard library filter encoded in JSON
+ * @return this builder
+ */
+ public ConsumerBuilder withServerSideFilter ( String filter ) {
+ fFilter = filter;
+ return this;
+ }
+
+ /**
+ * Build the consumer
+ * @return a consumer
+ */
+ public MRConsumer build ()
+ {
+ if ( fHosts == null || fHosts.size() == 0 || fTopic == null )
+ {
+ throw new IllegalArgumentException ( "You must provide at least one host and a topic name." );
+ }
+
+ if ( fGroup == null )
+ {
+ fGroup = UUID.randomUUID ().toString ();
+ fId = "0";
+ log.info ( "Creating non-restartable client with group " + fGroup + " and ID " + fId + "." );
+ }
+
+ if ( sfConsumerMock != null ) return sfConsumerMock;
+ try {
+ return new MRConsumerImpl ( fHosts, fTopic, fGroup, fId, fTimeoutMs, fLimit, fFilter, fApiKey, fApiSecret );
+ } catch (MalformedURLException e) {
+ throw new IllegalArgumentException(e);
+ }
+ }
+
+ private Collection<String> fHosts = null;
+ private String fTopic = null;
+ private String fGroup = null;
+ private String fId = null;
+ private String fApiKey = null;
+ private String fApiSecret = null;
+ private int fTimeoutMs = -1;
+ private int fLimit = -1;
+ private String fFilter = null;
+ }
+
+ /*************************************************************************/
+ /*************************************************************************/
+ /*************************************************************************/
+
+ /**
+ * A publisher builder
+ * @author author
+ */
+ public static class PublisherBuilder
+ {
+ public PublisherBuilder () {}
+
+ /**
+ * Set the MR/UEB host(s) to use
+ * @param hostlist The host(s) used in the URL to MR. Can be "host:port", can be multiple comma-separated entries.
+ * @return this builder
+ */
+ public PublisherBuilder usingHosts ( String hostlist ) {
+ return usingHosts ( MRConsumerImpl.stringToList(hostlist) );
+ }
+
+ /**
+ * Set the MR/UEB host(s) to use
+ * @param hostSet The host(s) used in the URL to MR. Can be "host:port"
+ * @return this builder
+ */
+ public PublisherBuilder usingHosts ( String[] hostSet )
+ {
+ final TreeSet<String> hosts = new TreeSet<String> ();
+ for ( String hp : hostSet )
+ {
+ hosts.add ( hp );
+ }
+ return usingHosts ( hosts );
+ }
+
+ /**
+ * Set the MR/UEB host(s) to use
+ * @param hostlist The host(s) used in the URL to MR. Can be "host:port".
+ * @return this builder
+ */
+ public PublisherBuilder usingHosts ( Collection<String> hostlist ) {
+ fHosts=hostlist;
+ return this;
+ }
+
+ /**
+ * Set the topic to publish on
+ * @param topic The topic on which to publish messages.
+ * @return this builder
+ */
+ public PublisherBuilder onTopic ( String topic ) {
+ fTopic = topic;
+ return this;
+ }
+
+ /**
+ * Batch message sends with the given limits.
+ * @param messageCount The largest set of messages to batch.
+ * @param ageInMs The maximum age of a message waiting in a batch.
+ * @return this builder
+ */
+ public PublisherBuilder limitBatch ( int messageCount, int ageInMs ) {
+ fMaxBatchSize = messageCount;
+ fMaxBatchAgeMs = ageInMs;
+ return this;
+ }
+
+ /**
+ * Compress transactions
+ * @return this builder
+ */
+ public PublisherBuilder withCompresion () {
+ return enableCompresion(true);
+ }
+
+ /**
+ * Do not compress transactions
+ * @return this builder
+ */
+ public PublisherBuilder withoutCompresion () {
+ return enableCompresion(false);
+ }
+
+ /**
+ * Set the compression option
+ * @param compress true to gzip compress transactions
+ * @return this builder
+ */
+ public PublisherBuilder enableCompresion ( boolean compress ) {
+ fCompress = compress;
+ return this;
+ }
+
+ /**
+ * Set the API key and secret for this client.
+ * @param apiKey
+ * @param apiSecret
+ * @return this builder
+ */
+ public PublisherBuilder authenticatedBy ( String apiKey, String apiSecret ) {
+ fApiKey = apiKey;
+ fApiSecret = apiSecret;
+ return this;
+ }
+
+ /**
+ * Build the publisher
+ * @return a batching publisher
+ */
+ public MRBatchingPublisher build ()
+ {
+ if ( fHosts == null || fHosts.isEmpty() || fTopic == null )
+ {
+ throw new IllegalArgumentException ( "You must provide at least one host and a topic name." );
+ }
+
+ if ( sfPublisherMock != null ) return sfPublisherMock;
+
+ final MRSimplerBatchPublisher pub = new MRSimplerBatchPublisher.Builder ().
+ againstUrls ( fHosts ).
+ onTopic ( fTopic ).
+ batchTo ( fMaxBatchSize, fMaxBatchAgeMs ).
+ compress ( fCompress ).
+ build ();
+ if ( fApiKey != null )
+ {
+ pub.setApiCredentials ( fApiKey, fApiSecret );
+ }
+ return pub;
+ }
+
+ private Collection<String> fHosts = null;
+ private String fTopic = null;
+ private int fMaxBatchSize = 1;
+ private int fMaxBatchAgeMs = 1;
+ private boolean fCompress = false;
+ private String fApiKey = null;
+ private String fApiSecret = null;
+ }
+
+ /**
+ * A builder for an identity manager
+ * @author author
+ */
+ public static class IdentityManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRIdentityManager>
+ {
+ /**
+ * Construct an identity manager builder.
+ */
+ public IdentityManagerBuilder () {}
+
+ @Override
+ protected MRIdentityManager constructClient ( Collection<String> hosts ) { try {
+ return new MRMetaClient ( hosts );
+ } catch (MalformedURLException e) {
+ throw new IllegalArgumentException(e);
+ } }
+ }
+
+ /**
+ * A builder for a topic manager
+ * @author author
+ */
+ public static class TopicManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRTopicManager>
+ {
+ /**
+ * Construct an topic manager builder.
+ */
+ public TopicManagerBuilder () {}
+
+ @Override
+ protected MRTopicManager constructClient ( Collection<String> hosts ) { try {
+ return new MRMetaClient ( hosts );
+ } catch (MalformedURLException e) {
+ throw new IllegalArgumentException(e);
+ } }
+ }
+
+ /**
+ * Inject a consumer. Used to support unit tests.
+ * @param cc
+ */
+ public static void $testInject ( MRConsumer cc )
+ {
+ sfConsumerMock = cc;
+ }
+
+ /**
+ * Inject a publisher. Used to support unit tests.
+ * @param pub
+ */
+ public static void $testInject ( MRBatchingPublisher pub )
+ {
+ sfPublisherMock = pub;
+ }
+
+ static MRConsumer sfConsumerMock = null;
+ static MRBatchingPublisher sfPublisherMock = null;
+
+ /**
+ * A builder for an identity manager
+ * @author author
+ */
+ public static abstract class AbstractAuthenticatedManagerBuilder<T extends MRClient>
+ {
+ /**
+ * Construct an identity manager builder.
+ */
+ public AbstractAuthenticatedManagerBuilder () {}
+
+ /**
+ * Set the host list
+ * @param hostList a comma-separated list of hosts to use to connect to MR
+ * @return this builder
+ */
+ public AbstractAuthenticatedManagerBuilder<T> usingHosts ( String hostList ) {
+ return usingHosts ( MRConsumerImpl.stringToList(hostList) );
+ }
+
+ /**
+ * Set the host list
+ * @param hostSet a set of hosts to use to connect to MR
+ * @return this builder
+ */
+ public AbstractAuthenticatedManagerBuilder<T> usingHosts ( Collection<String> hostSet ) {
+ fHosts = hostSet;
+ return this;
+ }
+
+ /**
+ * Set the API key and secret for this client.
+ * @param apiKey
+ * @param apiSecret
+ * @return this builder
+ */
+ public AbstractAuthenticatedManagerBuilder<T> authenticatedBy ( String apiKey, String apiSecret ) {
+ fApiKey = apiKey;
+ fApiSecret = apiSecret;
+ return this;
+ }
+
+ /**
+ * Build the consumer
+ * @return a consumer
+ */
+ public T build ()
+ {
+ if ( fHosts.isEmpty() )
+ {
+ throw new IllegalArgumentException ( "You must provide at least one host and a topic name." );
+ }
+
+ final T mgr = constructClient ( fHosts );
+ mgr.setApiCredentials ( fApiKey, fApiSecret );
+ return mgr;
+ }
+
+ protected abstract T constructClient ( Collection<String> hosts );
+
+ private Collection<String> fHosts = null;
+ private String fApiKey = null;
+ private String fApiSecret = null;
+ }
+
+ private static final Logger log = LoggerFactory.getLogger ( MRClientBuilders.class );
}
* ================================================================================
* Copyright © 2017 AT&T Intellectual Property. All rights reserved.
* ================================================================================
+ * Modifications Copyright © 2018 IBM.
+ * ================================================================================
* 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
* @author author
*/
public class MRClientFactory {
- public static MultivaluedMap<String, Object> HTTPHeadersMap;
- public static Map<String, String> DME2HeadersMap;
- public static String routeFilePath;
-
- public static FileReader routeReader;
-
- public static FileWriter routeWriter = null;
- public static Properties prop = null;
-
-
- // props= new Properties();
- /**
- * Create a consumer instance with the default timeout and no limit on
- * messages returned. This consumer operates as an independent consumer
- * (i.e., not in a group) and is NOT re-startable across sessions.
- *
- * @param hostList
- * A comma separated list of hosts to use to connect to MR. You
- * can include port numbers (3904 is the default). For example,
- * "hostname:8080,"
- *
- * @param topic
- * The topic to consume
- *
- * @return a consumer
- */
- public static MRConsumer createConsumer(String hostList, String topic) {
- return createConsumer(MRConsumerImpl.stringToList(hostList), topic);
- }
-
- /**
- * Create a consumer instance with the default timeout and no limit on
- * messages returned. This consumer operates as an independent consumer
- * (i.e., not in a group) and is NOT re-startable across sessions.
- *
- * @param hostSet
- * The host used in the URL to MR. Entries can be "host:port".
- * @param topic
- * The topic to consume
- *
- * @return a consumer
- */
- public static MRConsumer createConsumer(Collection<String> hostSet, String topic) {
- return createConsumer(hostSet, topic, null);
- }
-
- /**
- * Create a consumer instance with server-side filtering, the default
- * timeout, and no limit on messages returned. This consumer operates as an
- * independent consumer (i.e., not in a group) and is NOT re-startable
- * across sessions.
- *
- * @param hostSet
- * The host used in the URL to MR. Entries can be "host:port".
- * @param topic
- * The topic to consume
- * @param filter
- * a filter to use on the server side
- *
- * @return a consumer
- */
- public static MRConsumer createConsumer(Collection<String> hostSet, String topic, String filter) {
- return createConsumer(hostSet, topic, UUID.randomUUID().toString(), "0", -1, -1, filter, null, null);
- }
-
- /**
- * Create a consumer instance with the default timeout, and no limit on
- * messages returned. This consumer can operate in a logical group and is
- * re-startable across sessions when you use the same group and ID on
- * restart.
- *
- * @param hostSet
- * The host used in the URL to MR. Entries can be "host:port".
- * @param topic
- * The topic to consume
- * @param consumerGroup
- * The name of the consumer group this consumer is part of
- * @param consumerId
- * The unique id of this consume in its group
- *
- * @return a consumer
- */
- public static MRConsumer createConsumer(Collection<String> hostSet, final String topic, final String consumerGroup,
- final String consumerId) {
- return createConsumer(hostSet, topic, consumerGroup, consumerId, -1, -1);
- }
-
- /**
- * Create a consumer instance with the default timeout, and no limit on
- * messages returned. This consumer can operate in a logical group and is
- * re-startable across sessions when you use the same group and ID on
- * restart.
- *
- * @param hostSet
- * The host used in the URL to MR. Entries can be "host:port".
- * @param topic
- * The topic to consume
- * @param consumerGroup
- * The name of the consumer group this consumer is part of
- * @param consumerId
- * The unique id of this consume in its group
- * @param timeoutMs
- * The amount of time in milliseconds that the server should keep
- * the connection open while waiting for message traffic. Use -1
- * for default timeout.
- * @param limit
- * A limit on the number of messages returned in a single call.
- * Use -1 for no limit.
- *
- * @return a consumer
- */
- public static MRConsumer createConsumer(Collection<String> hostSet, final String topic, final String consumerGroup,
- final String consumerId, int timeoutMs, int limit) {
- return createConsumer(hostSet, topic, consumerGroup, consumerId, timeoutMs, limit, null, null, null);
- }
-
- /**
- * Create a consumer instance with the default timeout, and no limit on
- * messages returned. This consumer can operate in a logical group and is
- * re-startable across sessions when you use the same group and ID on
- * restart. This consumer also uses server-side filtering.
- *
- * @param hostList
- * A comma separated list of hosts to use to connect to MR. You
- * can include port numbers (3904 is the default). For example,
- * "ueb01hydc.it.att.com:8080,ueb02hydc.it.att.com"
- * @param topic
- * The topic to consume
- * @param consumerGroup
- * The name of the consumer group this consumer is part of
- * @param consumerId
- * The unique id of this consume in its group
- * @param timeoutMs
- * The amount of time in milliseconds that the server should keep
- * the connection open while waiting for message traffic. Use -1
- * for default timeout.
- * @param limit
- * A limit on the number of messages returned in a single call.
- * Use -1 for no limit.
- * @param filter
- * A Highland Park filter expression using only built-in filter
- * components. Use null for "no filter".
- *
- * @return a consumer
- */
- public static MRConsumer createConsumer(String hostList, final String topic, final String consumerGroup,
- final String consumerId, int timeoutMs, int limit, String filter, String apiKey, String apiSecret) {
- return createConsumer(MRConsumerImpl.stringToList(hostList), topic, consumerGroup, consumerId, timeoutMs, limit,
- filter, apiKey, apiSecret);
- }
-
- /**
- * Create a consumer instance with the default timeout, and no limit on
- * messages returned. This consumer can operate in a logical group and is
- * re-startable across sessions when you use the same group and ID on
- * restart. This consumer also uses server-side filtering.
- *
- * @param hostSet
- * The host used in the URL to MR. Entries can be "host:port".
- * @param topic
- * The topic to consume
- * @param consumerGroup
- * The name of the consumer group this consumer is part of
- * @param consumerId
- * The unique id of this consume in its group
- * @param timeoutMs
- * The amount of time in milliseconds that the server should keep
- * the connection open while waiting for message traffic. Use -1
- * for default timeout.
- * @param limit
- * A limit on the number of messages returned in a single call.
- * Use -1 for no limit.
- * @param filter
- * A Highland Park filter expression using only built-in filter
- * components. Use null for "no filter".
- *
- * @return a consumer
- */
- public static MRConsumer createConsumer(Collection<String> hostSet, final String topic, final String consumerGroup,
- final String consumerId, int timeoutMs, int limit, String filter, String apiKey, String apiSecret) {
- if (MRClientBuilders.sfConsumerMock != null)
- return MRClientBuilders.sfConsumerMock;
- try {
- return new MRConsumerImpl(hostSet, topic, consumerGroup, consumerId, timeoutMs, limit, filter, apiKey,
- apiSecret);
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException(e);
- }
- }
-
- /*************************************************************************/
- /*************************************************************************/
- /*************************************************************************/
-
- /**
- * Create a publisher that sends each message (or group of messages)
- * immediately. Most applications should favor higher latency for much
- * higher message throughput and the "simple publisher" is not a good
- * choice.
- *
- * @param hostlist
- * The host used in the URL to MR. Can be "host:port", can be
- * multiple comma-separated entries.
- * @param topic
- * The topic on which to publish messages.
- * @return a publisher
- */
- public static MRBatchingPublisher createSimplePublisher(String hostlist, String topic) {
- return createBatchingPublisher(hostlist, topic, 1, 1);
- }
-
- /**
- * Create a publisher that batches messages. Be sure to close the publisher
- * to send the last batch and ensure a clean shutdown. Message payloads are
- * not compressed.
- *
- * @param hostlist
- * The host used in the URL to MR. Can be "host:port", can be
- * multiple comma-separated entries.
- * @param topic
- * The topic on which to publish messages.
- * @param maxBatchSize
- * The largest set of messages to batch
- * @param maxAgeMs
- * The maximum age of a message waiting in a batch
- *
- * @return a publisher
- */
- public static MRBatchingPublisher createBatchingPublisher(String hostlist, String topic, int maxBatchSize,
- long maxAgeMs) {
- return createBatchingPublisher(hostlist, topic, maxBatchSize, maxAgeMs, false);
- }
-
- /**
- * Create a publisher that batches messages. Be sure to close the publisher
- * to send the last batch and ensure a clean shutdown.
- *
- * @param hostlist
- * The host used in the URL to MR. Can be "host:port", can be
- * multiple comma-separated entries.
- * @param topic
- * The topic on which to publish messages.
- * @param maxBatchSize
- * The largest set of messages to batch
- * @param maxAgeMs
- * The maximum age of a message waiting in a batch
- * @param compress
- * use gzip compression
- *
- * @return a publisher
- */
- public static MRBatchingPublisher createBatchingPublisher(String hostlist, String topic, int maxBatchSize,
- long maxAgeMs, boolean compress) {
- return createBatchingPublisher(MRConsumerImpl.stringToList(hostlist), topic, maxBatchSize, maxAgeMs, compress);
- }
-
- /**
- * Create a publisher that batches messages. Be sure to close the publisher
- * to send the last batch and ensure a clean shutdown.
- *
- * @param hostSet
- * A set of hosts to be used in the URL to MR. Can be
- * "host:port". Use multiple entries to enable failover.
- * @param topic
- * The topic on which to publish messages.
- * @param maxBatchSize
- * The largest set of messages to batch
- * @param maxAgeMs
- * The maximum age of a message waiting in a batch
- * @param compress
- * use gzip compression
- *
- * @return a publisher
- */
- public static MRBatchingPublisher createBatchingPublisher(String[] hostSet, String topic, int maxBatchSize,
- long maxAgeMs, boolean compress) {
- final TreeSet<String> hosts = new TreeSet<String>();
- for (String hp : hostSet) {
- hosts.add(hp);
- }
- return createBatchingPublisher(hosts, topic, maxBatchSize, maxAgeMs, compress);
- }
-
- /**
- * Create a publisher that batches messages. Be sure to close the publisher
- * to send the last batch and ensure a clean shutdown.
- *
- * @param hostSet
- * A set of hosts to be used in the URL to MR. Can be
- * "host:port". Use multiple entries to enable failover.
- * @param topic
- * The topic on which to publish messages.
- * @param maxBatchSize
- * The largest set of messages to batch
- * @param maxAgeMs
- * The maximum age of a message waiting in a batch
- * @param compress
- * use gzip compression
- *
- * @return a publisher
- */
- public static MRBatchingPublisher createBatchingPublisher(Collection<String> hostSet, String topic,
- int maxBatchSize, long maxAgeMs, boolean compress) {
- return new MRSimplerBatchPublisher.Builder().againstUrls(hostSet).onTopic(topic).batchTo(maxBatchSize, maxAgeMs)
- .compress(compress).build();
- }
-
- /**
- * Create a publisher that batches messages. Be sure to close the publisher
- * to send the last batch and ensure a clean shutdown.
- *
- * @param host
- * A host to be used in the URL to MR. Can be "host:port". Use
- * multiple entries to enable failover.
- * @param topic
- * The topic on which to publish messages.
- * @param username
- * username
- * @param password
- * password
- * @param maxBatchSize
- * The largest set of messages to batch
- * @param maxAgeMs
- * The maximum age of a message waiting in a batch
- * @param compress
- * use gzip compression
- * @param protocolFlag
- * http auth or ueb auth or dme2 method
- * @param producerFilePath
- * all properties for publisher
- * @return MRBatchingPublisher obj
- */
- public static MRBatchingPublisher createBatchingPublisher(String host, String topic, final String username,
- final String password, int maxBatchSize, long maxAgeMs, boolean compress, String protocolFlag,
- String producerFilePath) {
- MRSimplerBatchPublisher pub = new MRSimplerBatchPublisher.Builder()
- .againstUrls(MRConsumerImpl.stringToList(host)).onTopic(topic).batchTo(maxBatchSize, maxAgeMs)
- .compress(compress).build();
-
- pub.setHost(host);
- pub.setUsername(username);
- pub.setPassword(password);
- pub.setProtocolFlag(protocolFlag);
- 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 Properties
- * props set all properties for publishing message
- * @return MRBatchingPublisher obj
- * @throws FileNotFoundException
- * exc
- * @throws IOException
- * ioex
- */
- public static MRBatchingPublisher createBatchingPublisher(Properties props, boolean withResponse)
- throws FileNotFoundException, IOException {
- return createInternalBatchingPublisher(props, withResponse);
- }
-
- /**
- * Create a publisher that batches messages. Be sure to close the publisher
- * to send the last batch and ensure a clean shutdown
- *
- * @param Properties
- * props set all properties for publishing message
- * @return MRBatchingPublisher obj
- * @throws FileNotFoundException
- * exc
- * @throws IOException
- * ioex
- */
- public static MRBatchingPublisher createBatchingPublisher(Properties props)
- throws FileNotFoundException, IOException {
- return createInternalBatchingPublisher(props, false);
- }
-
- /**
- * 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);
- return createBatchingPublisher(props);
- }
-
- /**
- * 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);
- return createBatchingPublisher(props, withResponse);
- }
-
- protected static MRBatchingPublisher createInternalBatchingPublisher(Properties props, boolean withResponse)
- throws FileNotFoundException, IOException {
- assert props != null;
- MRSimplerBatchPublisher pub;
- if (withResponse) {
- pub = new MRSimplerBatchPublisher.Builder()
- .againstUrlsOrServiceName(MRConsumerImpl.stringToList(props.getProperty("host")),MRConsumerImpl.stringToList(props.getProperty("ServiceName")), props.getProperty("TransportType"))
- .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();
- } else {
- pub = new MRSimplerBatchPublisher.Builder()
- .againstUrlsOrServiceName(MRConsumerImpl.stringToList(props.getProperty("host")), MRConsumerImpl.stringToList(props.getProperty("ServiceName")), props.getProperty("TransportType"))
- .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.setProtocolFlag(props.getProperty("TransportType"));
- pub.setProps(props);
- prop = new Properties();
- if (props.getProperty("TransportType").equalsIgnoreCase(ProtocolTypeConstants.DME2.getValue())) {
- routeFilePath = props.getProperty("DME2preferredRouterFilePath");
- routeReader = new FileReader(new File(routeFilePath));
- File fo = new File(routeFilePath);
- if (!fo.exists()) {
- routeWriter = new FileWriter(new File(routeFilePath));
- }
- }
- return pub;
- }
-
- /**
- * Create an identity manager client to work with API keys.
- *
- * @param hostSet
- * A set of hosts to be used in the URL to MR. Can be
- * "host:port". Use multiple entries to enable failover.
- * @param apiKey
- * Your API key
- * @param apiSecret
- * Your API secret
- * @return an identity manager
- */
- public static MRIdentityManager createIdentityManager(Collection<String> hostSet, String apiKey, String apiSecret) {
- MRIdentityManager cim;
- try {
- cim = new MRMetaClient(hostSet);
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException(e);
- }
- cim.setApiCredentials(apiKey, apiSecret);
- return cim;
- }
-
- /**
- * Create a topic manager for working with topics.
- *
- * @param hostSet
- * A set of hosts to be used in the URL to MR. Can be
- * "host:port". Use multiple entries to enable failover.
- * @param apiKey
- * Your API key
- * @param apiSecret
- * Your API secret
- * @return a topic manager
- */
- public static MRTopicManager createTopicManager(Collection<String> hostSet, String apiKey, String apiSecret) {
- MRMetaClient tmi;
- try {
- tmi = new MRMetaClient(hostSet);
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException(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 IllegalArgumentException(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 IllegalArgumentException(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);
-
- return createConsumer(props);
- }
-
- public static MRConsumer createConsumer(Properties props) throws FileNotFoundException, IOException {
- int timeout;
- ValidatorUtil.validateSubscriber(props);
- 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.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;
- }
+ private static final String AUTH_KEY = "authKey";
+ private static final String AUTH_DATE = "authDate";
+ private static final String PASSWORD = "password";
+ private static final String USERNAME = "username";
+ private static final String DME2PREFERRED_ROUTER_FILE_PATH = "DME2preferredRouterFilePath";
+ private static final String TOPIC = "topic";
+ private static final String TRANSPORT_TYPE = "TransportType";
+ public static MultivaluedMap<String, Object> HTTPHeadersMap;
+ public static Map<String, String> DME2HeadersMap;
+ public static String routeFilePath;
+
+ public static FileReader routeReader;
+
+ public static FileWriter routeWriter = null;
+ public static Properties prop = null;
+
+ /**
+ * Instantiates MRClientFactory.
+ */
+ private MRClientFactory() {
+ //prevents instantiation.
+ }
+
+ /**
+ * Create a consumer instance with the default timeout and no limit on
+ * messages returned. This consumer operates as an independent consumer
+ * (i.e., not in a group) and is NOT re-startable across sessions.
+ *
+ * @param hostList
+ * A comma separated list of hosts to use to connect to MR. You
+ * can include port numbers (3904 is the default). For example,
+ * "hostname:8080,"
+ *
+ * @param topic
+ * The topic to consume
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer(String hostList, String topic) {
+ return createConsumer(MRConsumerImpl.stringToList(hostList), topic);
+ }
+
+ /**
+ * Create a consumer instance with the default timeout and no limit on
+ * messages returned. This consumer operates as an independent consumer
+ * (i.e., not in a group) and is NOT re-startable across sessions.
+ *
+ * @param hostSet
+ * The host used in the URL to MR. Entries can be "host:port".
+ * @param topic
+ * The topic to consume
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer(Collection<String> hostSet, String topic) {
+ return createConsumer(hostSet, topic, null);
+ }
+
+ /**
+ * Create a consumer instance with server-side filtering, the default
+ * timeout, and no limit on messages returned. This consumer operates as an
+ * independent consumer (i.e., not in a group) and is NOT re-startable
+ * across sessions.
+ *
+ * @param hostSet
+ * The host used in the URL to MR. Entries can be "host:port".
+ * @param topic
+ * The topic to consume
+ * @param filter
+ * a filter to use on the server side
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer(Collection<String> hostSet, String topic, String filter) {
+ return createConsumer(hostSet, topic, UUID.randomUUID().toString(), "0", -1, -1, filter, null, null);
+ }
+
+ /**
+ * Create a consumer instance with the default timeout, and no limit on
+ * messages returned. This consumer can operate in a logical group and is
+ * re-startable across sessions when you use the same group and ID on
+ * restart.
+ *
+ * @param hostSet
+ * The host used in the URL to MR. Entries can be "host:port".
+ * @param topic
+ * The topic to consume
+ * @param consumerGroup
+ * The name of the consumer group this consumer is part of
+ * @param consumerId
+ * The unique id of this consume in its group
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer(Collection<String> hostSet, final String topic, final String consumerGroup,
+ final String consumerId) {
+ return createConsumer(hostSet, topic, consumerGroup, consumerId, -1, -1);
+ }
+
+ /**
+ * Create a consumer instance with the default timeout, and no limit on
+ * messages returned. This consumer can operate in a logical group and is
+ * re-startable across sessions when you use the same group and ID on
+ * restart.
+ *
+ * @param hostSet
+ * The host used in the URL to MR. Entries can be "host:port".
+ * @param topic
+ * The topic to consume
+ * @param consumerGroup
+ * The name of the consumer group this consumer is part of
+ * @param consumerId
+ * The unique id of this consume in its group
+ * @param timeoutMs
+ * The amount of time in milliseconds that the server should keep
+ * the connection open while waiting for message traffic. Use -1
+ * for default timeout.
+ * @param limit
+ * A limit on the number of messages returned in a single call.
+ * Use -1 for no limit.
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer(Collection<String> hostSet, final String topic, final String consumerGroup,
+ final String consumerId, int timeoutMs, int limit) {
+ return createConsumer(hostSet, topic, consumerGroup, consumerId, timeoutMs, limit, null, null, null);
+ }
+
+ /**
+ * Create a consumer instance with the default timeout, and no limit on
+ * messages returned. This consumer can operate in a logical group and is
+ * re-startable across sessions when you use the same group and ID on
+ * restart. This consumer also uses server-side filtering.
+ *
+ * @param hostList
+ * A comma separated list of hosts to use to connect to MR. You
+ * can include port numbers (3904 is the default). For example,
+ * "ueb01hydc.it.att.com:8080,ueb02hydc.it.att.com"
+ * @param topic
+ * The topic to consume
+ * @param consumerGroup
+ * The name of the consumer group this consumer is part of
+ * @param consumerId
+ * The unique id of this consume in its group
+ * @param timeoutMs
+ * The amount of time in milliseconds that the server should keep
+ * the connection open while waiting for message traffic. Use -1
+ * for default timeout.
+ * @param limit
+ * A limit on the number of messages returned in a single call.
+ * Use -1 for no limit.
+ * @param filter
+ * A Highland Park filter expression using only built-in filter
+ * components. Use null for "no filter".
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer(String hostList, final String topic, final String consumerGroup,
+ final String consumerId, int timeoutMs, int limit, String filter, String apiKey, String apiSecret) {
+ return createConsumer(MRConsumerImpl.stringToList(hostList), topic, consumerGroup, consumerId, timeoutMs, limit,
+ filter, apiKey, apiSecret);
+ }
+
+ /**
+ * Create a consumer instance with the default timeout, and no limit on
+ * messages returned. This consumer can operate in a logical group and is
+ * re-startable across sessions when you use the same group and ID on
+ * restart. This consumer also uses server-side filtering.
+ *
+ * @param hostSet
+ * The host used in the URL to MR. Entries can be "host:port".
+ * @param topic
+ * The topic to consume
+ * @param consumerGroup
+ * The name of the consumer group this consumer is part of
+ * @param consumerId
+ * The unique id of this consume in its group
+ * @param timeoutMs
+ * The amount of time in milliseconds that the server should keep
+ * the connection open while waiting for message traffic. Use -1
+ * for default timeout.
+ * @param limit
+ * A limit on the number of messages returned in a single call.
+ * Use -1 for no limit.
+ * @param filter
+ * A Highland Park filter expression using only built-in filter
+ * components. Use null for "no filter".
+ *
+ * @return a consumer
+ */
+ public static MRConsumer createConsumer(Collection<String> hostSet, final String topic, final String consumerGroup,
+ final String consumerId, int timeoutMs, int limit, String filter, String apiKey, String apiSecret) {
+ if (MRClientBuilders.sfConsumerMock != null)
+ return MRClientBuilders.sfConsumerMock;
+ try {
+ return new MRConsumerImpl(hostSet, topic, consumerGroup, consumerId, timeoutMs, limit, filter, apiKey,
+ apiSecret);
+ } catch (MalformedURLException e) {
+ throw new IllegalArgumentException(e);
+ }
+ }
+
+ /*************************************************************************/
+ /*************************************************************************/
+ /*************************************************************************/
+
+ /**
+ * Create a publisher that sends each message (or group of messages)
+ * immediately. Most applications should favor higher latency for much
+ * higher message throughput and the "simple publisher" is not a good
+ * choice.
+ *
+ * @param hostlist
+ * The host used in the URL to MR. Can be "host:port", can be
+ * multiple comma-separated entries.
+ * @param topic
+ * The topic on which to publish messages.
+ * @return a publisher
+ */
+ public static MRBatchingPublisher createSimplePublisher(String hostlist, String topic) {
+ return createBatchingPublisher(hostlist, topic, 1, 1);
+ }
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher
+ * to send the last batch and ensure a clean shutdown. Message payloads are
+ * not compressed.
+ *
+ * @param hostlist
+ * The host used in the URL to MR. Can be "host:port", can be
+ * multiple comma-separated entries.
+ * @param topic
+ * The topic on which to publish messages.
+ * @param maxBatchSize
+ * The largest set of messages to batch
+ * @param maxAgeMs
+ * The maximum age of a message waiting in a batch
+ *
+ * @return a publisher
+ */
+ public static MRBatchingPublisher createBatchingPublisher(String hostlist, String topic, int maxBatchSize,
+ long maxAgeMs) {
+ return createBatchingPublisher(hostlist, topic, maxBatchSize, maxAgeMs, false);
+ }
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher
+ * to send the last batch and ensure a clean shutdown.
+ *
+ * @param hostlist
+ * The host used in the URL to MR. Can be "host:port", can be
+ * multiple comma-separated entries.
+ * @param topic
+ * The topic on which to publish messages.
+ * @param maxBatchSize
+ * The largest set of messages to batch
+ * @param maxAgeMs
+ * The maximum age of a message waiting in a batch
+ * @param compress
+ * use gzip compression
+ *
+ * @return a publisher
+ */
+ public static MRBatchingPublisher createBatchingPublisher(String hostlist, String topic, int maxBatchSize,
+ long maxAgeMs, boolean compress) {
+ return createBatchingPublisher(MRConsumerImpl.stringToList(hostlist), topic, maxBatchSize, maxAgeMs, compress);
+ }
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher
+ * to send the last batch and ensure a clean shutdown.
+ *
+ * @param hostSet
+ * A set of hosts to be used in the URL to MR. Can be
+ * "host:port". Use multiple entries to enable failover.
+ * @param topic
+ * The topic on which to publish messages.
+ * @param maxBatchSize
+ * The largest set of messages to batch
+ * @param maxAgeMs
+ * The maximum age of a message waiting in a batch
+ * @param compress
+ * use gzip compression
+ *
+ * @return a publisher
+ */
+ public static MRBatchingPublisher createBatchingPublisher(String[] hostSet, String topic, int maxBatchSize,
+ long maxAgeMs, boolean compress) {
+ final TreeSet<String> hosts = new TreeSet<String>();
+ for (String hp : hostSet) {
+ hosts.add(hp);
+ }
+ return createBatchingPublisher(hosts, topic, maxBatchSize, maxAgeMs, compress);
+ }
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher
+ * to send the last batch and ensure a clean shutdown.
+ *
+ * @param hostSet
+ * A set of hosts to be used in the URL to MR. Can be
+ * "host:port". Use multiple entries to enable failover.
+ * @param topic
+ * The topic on which to publish messages.
+ * @param maxBatchSize
+ * The largest set of messages to batch
+ * @param maxAgeMs
+ * The maximum age of a message waiting in a batch
+ * @param compress
+ * use gzip compression
+ *
+ * @return a publisher
+ */
+ public static MRBatchingPublisher createBatchingPublisher(Collection<String> hostSet, String topic,
+ int maxBatchSize, long maxAgeMs, boolean compress) {
+ return new MRSimplerBatchPublisher.Builder().againstUrls(hostSet).onTopic(topic).batchTo(maxBatchSize, maxAgeMs)
+ .compress(compress).build();
+ }
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher
+ * to send the last batch and ensure a clean shutdown.
+ *
+ * @param host
+ * A host to be used in the URL to MR. Can be "host:port". Use
+ * multiple entries to enable failover.
+ * @param topic
+ * The topic on which to publish messages.
+ * @param username
+ * username
+ * @param password
+ * password
+ * @param maxBatchSize
+ * The largest set of messages to batch
+ * @param maxAgeMs
+ * The maximum age of a message waiting in a batch
+ * @param compress
+ * use gzip compression
+ * @param protocolFlag
+ * http auth or ueb auth or dme2 method
+ * @param producerFilePath
+ * all properties for publisher
+ * @return MRBatchingPublisher obj
+ */
+ public static MRBatchingPublisher createBatchingPublisher(String host, String topic, final String username,
+ final String password, int maxBatchSize, long maxAgeMs, boolean compress, String protocolFlag) {
+ 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);
+ 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 Properties
+ * props set all properties for publishing message
+ * @return MRBatchingPublisher obj
+ * @throws FileNotFoundException
+ * exc
+ * @throws IOException
+ * ioex
+ */
+ public static MRBatchingPublisher createBatchingPublisher(Properties props, boolean withResponse)
+ throws FileNotFoundException, IOException {
+ return createInternalBatchingPublisher(props, withResponse);
+ }
+
+ /**
+ * Create a publisher that batches messages. Be sure to close the publisher
+ * to send the last batch and ensure a clean shutdown
+ *
+ * @param Properties
+ * props set all properties for publishing message
+ * @return MRBatchingPublisher obj
+ * @throws FileNotFoundException
+ * exc
+ * @throws IOException
+ * ioex
+ */
+ public static MRBatchingPublisher createBatchingPublisher(Properties props)
+ throws FileNotFoundException, IOException {
+ return createInternalBatchingPublisher(props, false);
+ }
+
+ /**
+ * 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);
+ return createBatchingPublisher(props);
+ }
+
+ /**
+ * 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);
+ return createBatchingPublisher(props, withResponse);
+ }
+
+ protected static MRBatchingPublisher createInternalBatchingPublisher(Properties props, boolean withResponse)
+ throws FileNotFoundException, IOException {
+ assert props != null;
+ MRSimplerBatchPublisher pub;
+ if (withResponse) {
+ pub = new MRSimplerBatchPublisher.Builder()
+ .againstUrlsOrServiceName(MRConsumerImpl.stringToList(props.getProperty("host")),MRConsumerImpl.stringToList(props.getProperty("ServiceName")), props.getProperty(TRANSPORT_TYPE))
+ .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();
+ } else {
+ pub = new MRSimplerBatchPublisher.Builder()
+ .againstUrlsOrServiceName(MRConsumerImpl.stringToList(props.getProperty("host")), MRConsumerImpl.stringToList(props.getProperty("ServiceName")), props.getProperty(TRANSPORT_TYPE))
+ .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(TRANSPORT_TYPE).equalsIgnoreCase(ProtocolTypeConstants.AUTH_KEY.getValue())) {
+
+ pub.setAuthKey(props.getProperty(AUTH_KEY));
+ pub.setAuthDate(props.getProperty(AUTH_DATE));
+ pub.setUsername(props.getProperty(USERNAME));
+ pub.setPassword(props.getProperty(PASSWORD));
+ } else {
+ pub.setUsername(props.getProperty(USERNAME));
+ pub.setPassword(props.getProperty(PASSWORD));
+ }
+ pub.setProtocolFlag(props.getProperty(TRANSPORT_TYPE));
+ pub.setProps(props);
+ prop = new Properties();
+ if (props.getProperty(TRANSPORT_TYPE).equalsIgnoreCase(ProtocolTypeConstants.DME2.getValue())) {
+ routeFilePath = props.getProperty(DME2PREFERRED_ROUTER_FILE_PATH);
+ routeReader = new FileReader(new File(routeFilePath));
+ File fo = new File(routeFilePath);
+ if (!fo.exists()) {
+ routeWriter = new FileWriter(new File(routeFilePath));
+ }
+ }
+ return pub;
+ }
+
+ /**
+ * Create an identity manager client to work with API keys.
+ *
+ * @param hostSet
+ * A set of hosts to be used in the URL to MR. Can be
+ * "host:port". Use multiple entries to enable failover.
+ * @param apiKey
+ * Your API key
+ * @param apiSecret
+ * Your API secret
+ * @return an identity manager
+ */
+ public static MRIdentityManager createIdentityManager(Collection<String> hostSet, String apiKey, String apiSecret) {
+ MRIdentityManager cim;
+ try {
+ cim = new MRMetaClient(hostSet);
+ } catch (MalformedURLException e) {
+ throw new IllegalArgumentException(e);
+ }
+ cim.setApiCredentials(apiKey, apiSecret);
+ return cim;
+ }
+
+ /**
+ * Create a topic manager for working with topics.
+ *
+ * @param hostSet
+ * A set of hosts to be used in the URL to MR. Can be
+ * "host:port". Use multiple entries to enable failover.
+ * @param apiKey
+ * Your API key
+ * @param apiSecret
+ * Your API secret
+ * @return a topic manager
+ */
+ public static MRTopicManager createTopicManager(Collection<String> hostSet, String apiKey, String apiSecret) {
+ MRMetaClient tmi;
+ try {
+ tmi = new MRMetaClient(hostSet);
+ } catch (MalformedURLException e) {
+ throw new IllegalArgumentException(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 IllegalArgumentException(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 IllegalArgumentException(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);
+
+ return createConsumer(props);
+ }
+
+ public static MRConsumer createConsumer(Properties props) throws FileNotFoundException, IOException {
+ int timeout;
+ ValidatorUtil.validateSubscriber(props);
+ 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(TRANSPORT_TYPE).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(AUTH_KEY), props.getProperty(AUTH_DATE));
+ sub.setAuthKey(props.getProperty(AUTH_KEY));
+ sub.setAuthDate(props.getProperty(AUTH_DATE));
+ 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.setProps(props);
+ sub.setHost(props.getProperty("host"));
+ sub.setProtocolFlag(props.getProperty(TRANSPORT_TYPE));
+ sub.setfFilter(props.getProperty("filter"));
+ if (props.getProperty(TRANSPORT_TYPE).equalsIgnoreCase(ProtocolTypeConstants.DME2.getValue())) {
+ MRConsumerImpl.setRouterFilePath(props.getProperty(DME2PREFERRED_ROUTER_FILE_PATH));
+ routeFilePath = props.getProperty(DME2PREFERRED_ROUTER_FILE_PATH);
+ 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;
+ }
}
* ================================================================================\r
* Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.\r
* ================================================================================\r
+ * Modifications Copyright © 2018 IBM.\r
+ * ================================================================================\r
* Licensed under the Apache License, Version 2.0 (the "License");\r
* you may not use this file except in compliance with the License.\r
* You may obtain a copy of the License at\r
\r
public class MRClientFactoryTest {\r
\r
- private Collection<String> hostSet = new ArrayList<String>();\r
-\r
- private MRClientFactory factory = null;\r
-\r
- private String[] hostArray = new String[10];\r
-\r
- @Before\r
- public void setUp() throws Exception {\r
+ private Collection<String> hostSet = new ArrayList<String>();\r
\r
- for (int i = 0; i < 10; i++) {\r
- hostSet.add("host" + (i + 1));\r
- hostArray[i] = "host" + (i + 1);\r
- }\r
+ private String[] hostArray = new String[10];\r
\r
- factory = new MRClientFactory();\r
+ @Before\r
+ public void setUp() throws Exception {\r
\r
- }\r
+ for (int i = 0; i < 10; i++) {\r
+ hostSet.add("host" + (i + 1));\r
+ hostArray[i] = "host" + (i + 1);\r
+ }\r
+ }\r
\r
- @After\r
- public void tearDown() throws Exception {\r
+ @After\r
+ public void tearDown() throws Exception {\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateConsumer() {\r
+ @Test\r
+ public void testCreateConsumer() {\r
\r
- MRClientFactory.createConsumer("hostList hostList2", "testTopic");\r
- assertTrue(true);\r
+ MRClientFactory.createConsumer("hostList hostList2", "testTopic");\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateConsumer2() {\r
+ @Test\r
+ public void testCreateConsumer2() {\r
\r
- MRClientFactory.createConsumer(hostSet, "testTopic");\r
- assertTrue(true);\r
+ MRClientFactory.createConsumer(hostSet, "testTopic");\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateConsumer3() {\r
+ @Test\r
+ public void testCreateConsumer3() {\r
\r
- MRClientFactory.createConsumer(hostSet, "testTopic", "filter");\r
- assertTrue(true);\r
+ MRClientFactory.createConsumer(hostSet, "testTopic", "filter");\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateConsumer4() {\r
+ @Test\r
+ public void testCreateConsumer4() {\r
\r
- MRClientFactory.createConsumer(hostSet, "testTopic", "CG1", "22");\r
- assertTrue(true);\r
+ MRClientFactory.createConsumer(hostSet, "testTopic", "CG1", "22");\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateConsumer5() {\r
+ @Test\r
+ public void testCreateConsumer5() {\r
\r
- MRClientFactory.createConsumer(hostSet, "testTopic", "CG1", "22", 100, 100);\r
- assertTrue(true);\r
+ MRClientFactory.createConsumer(hostSet, "testTopic", "CG1", "22", 100, 100);\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateConsumer6() {\r
+ @Test\r
+ public void testCreateConsumer6() {\r
\r
- MRClientFactory.createConsumer("hostList", "testTopic", "CG1", "22", 100, 100, "filter", "apikey", "apisecret");\r
- assertTrue(true);\r
+ MRClientFactory.createConsumer("hostList", "testTopic", "CG1", "22", 100, 100, "filter", "apikey", "apisecret");\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateConsumer7() {\r
+ @Test\r
+ public void testCreateConsumer7() {\r
\r
- MRClientFactory.createConsumer(hostSet, "testTopic", "CG1", "22", 100, 100, "filter", "apikey", "apisecret");\r
- assertTrue(true);\r
+ MRClientFactory.createConsumer(hostSet, "testTopic", "CG1", "22", 100, 100, "filter", "apikey", "apisecret");\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateSimplePublisher() {\r
+ @Test\r
+ public void testCreateSimplePublisher() {\r
\r
- MRClientFactory.createSimplePublisher("hostList", "testTopic");\r
- assertTrue(true);\r
+ MRClientFactory.createSimplePublisher("hostList", "testTopic");\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateBatchingPublisher1() {\r
+ @Test\r
+ public void testCreateBatchingPublisher1() {\r
\r
- MRClientFactory.createBatchingPublisher("hostList", "testTopic", 100, 10);\r
- assertTrue(true);\r
+ MRClientFactory.createBatchingPublisher("hostList", "testTopic", 100, 10);\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateBatchingPublisher2() {\r
+ @Test\r
+ public void testCreateBatchingPublisher2() {\r
\r
- MRClientFactory.createBatchingPublisher("hostList", "testTopic", 100, 10, true);\r
- assertTrue(true);\r
+ MRClientFactory.createBatchingPublisher("hostList", "testTopic", 100, 10, true);\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateBatchingPublisher3() {\r
+ @Test\r
+ public void testCreateBatchingPublisher3() {\r
\r
- MRClientFactory.createBatchingPublisher(hostArray, "testTopic", 100, 10, true);\r
- assertTrue(true);\r
+ MRClientFactory.createBatchingPublisher(hostArray, "testTopic", 100, 10, true);\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateBatchingPublisher4() {\r
+ @Test\r
+ public void testCreateBatchingPublisher4() {\r
\r
- MRClientFactory.createBatchingPublisher(hostSet, "testTopic", 100, 10, true);\r
- assertTrue(true);\r
+ MRClientFactory.createBatchingPublisher(hostSet, "testTopic", 100, 10, true);\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateBatchingPublisher5() {\r
+ @Test\r
+ public void testCreateBatchingPublisher5() {\r
\r
- MRClientFactory.createBatchingPublisher("host", "testTopic", "username", "password", 100, 10, true,\r
- "protocolFlag", "/producer");\r
- assertTrue(true);\r
+ MRClientFactory.createBatchingPublisher("host", "testTopic", "username", "password", 100, 10, true,\r
+ "protocolFlag");\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateBatchingPublisher6() {\r
+ @Test\r
+ public void testCreateBatchingPublisher6() {\r
\r
- try {\r
- MRClientFactory.createBatchingPublisher("/producer");\r
- } catch (IOException e) {\r
- // TODO Auto-generated catch block\r
- e.printStackTrace();\r
- }\r
- assertTrue(true);\r
+ try {\r
+ MRClientFactory.createBatchingPublisher("/producer");\r
+ } catch (IOException e) {\r
+ // TODO Auto-generated catch block\r
+ e.printStackTrace();\r
+ }\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateBatchingPublisher7() {\r
+ @Test\r
+ public void testCreateBatchingPublisher7() {\r
\r
- try {\r
- MRClientFactory.createBatchingPublisher("/producer", true);\r
- } catch (IOException e) {\r
- // TODO Auto-generated catch block\r
- e.printStackTrace();\r
- }\r
- assertTrue(true);\r
+ try {\r
+ MRClientFactory.createBatchingPublisher("/producer", true);\r
+ } catch (IOException e) {\r
+ // TODO Auto-generated catch block\r
+ e.printStackTrace();\r
+ }\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateIdentityManager() {\r
+ @Test\r
+ public void testCreateIdentityManager() {\r
\r
- MRClientFactory.createIdentityManager(hostSet, "apikey", "apisecret");\r
+ MRClientFactory.createIdentityManager(hostSet, "apikey", "apisecret");\r
\r
- assertTrue(true);\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateTopicManager() {\r
+ @Test\r
+ public void testCreateTopicManager() {\r
\r
- MRClientFactory.createTopicManager(hostSet, "apikey", "apisecret");\r
+ MRClientFactory.createTopicManager(hostSet, "apikey", "apisecret");\r
\r
- assertTrue(true);\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateConsumer8() {\r
+ @Test\r
+ public void testCreateConsumer8() {\r
\r
- try {\r
- MRClientFactory.createConsumer("/consumer");\r
- } catch (IOException e) {\r
- // TODO Auto-generated catch block\r
- e.printStackTrace();\r
- }\r
+ try {\r
+ MRClientFactory.createConsumer("/consumer");\r
+ } catch (IOException e) {\r
+ // TODO Auto-generated catch block\r
+ e.printStackTrace();\r
+ }\r
\r
- assertTrue(true);\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateConsumer9() {\r
+ @Test\r
+ public void testCreateConsumer9() {\r
\r
- MRClientFactory.createConsumer("host", "topic", "username", "password", "group", "23", "protocolFlag",\r
- "/consumer", 1, 2);\r
+ MRClientFactory.createConsumer("host", "topic", "username", "password", "group", "23", "protocolFlag",\r
+ "/consumer", 1, 2);\r
\r
- assertTrue(true);\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
- @Test\r
- public void testCreateConsumer10() {\r
+ @Test\r
+ public void testCreateConsumer10() {\r
\r
- MRClientFactory.createConsumer("host", "topic", "username", "password", "group", "23", 1, 2, "protocolFlag",\r
- "/consumer");\r
+ MRClientFactory.createConsumer("host", "topic", "username", "password", "group", "23", 1, 2, "protocolFlag",\r
+ "/consumer");\r
\r
- assertTrue(true);\r
+ assertTrue(true);\r
\r
- }\r
- \r
- @Test\r
- public void test$testInject() {\r
+ }\r
+ \r
+ @Test\r
+ public void test$testInject() {\r
\r
- MRClientFactory.$testInject(null);\r
- assertTrue(true);\r
+ MRClientFactory.$testInject(null);\r
+ assertTrue(true);\r
\r
- }\r
+ }\r
\r
}
\ No newline at end of file
Code Review / dmaap / messagerouter / dmaapclient.git / commitdiff