src/main/java/org/onap/dmaap/mr/client/MRClientBuilders.java

   1 /*******************************************************************************
   2  *  ============LICENSE_START=======================================================
   3  *  org.onap.dmaap
   4  *  ================================================================================
   5  *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  *  ================================================================================
   7  *   Modifications Copyright © 2018 IBM.
   8  *  ================================================================================
   9  *  Licensed under the Apache License, Version 2.0 (the "License");
  10  *  you may not use this file except in compliance with the License.
  11  *  You may obtain a copy of the License at
  12  *        http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  *  Unless required by applicable law or agreed to in writing, software
  15  *  distributed under the License is distributed on an "AS IS" BASIS,
  16  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  *  See the License for the specific language governing permissions and
  18  *  limitations under the License.
  19  *  ============LICENSE_END=========================================================
  20  *
  21  *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
  22  *
  23  *******************************************************************************/
  24 package org.onap.dmaap.mr.client;
  25
  26 import java.net.MalformedURLException;
  27 import java.util.Collection;
  28 import java.util.TreeSet;
  29 import java.util.UUID;
  30
  31 import org.slf4j.Logger;
  32 import org.slf4j.LoggerFactory;
  33
  34 import org.onap.dmaap.mr.client.impl.MRConsumerImpl;
  35 import org.onap.dmaap.mr.client.impl.MRMetaClient;
  36 import org.onap.dmaap.mr.client.impl.MRSimplerBatchPublisher;
  37
  38 /**
  39  * A collection of builders for various types of MR API clients
  40  *
  41  * @author author
  42  */
  43 public class MRClientBuilders
  44 {
  45     private final static String ILLEGAL_ARGUMENT_MESSAGE = "You must provide at least one host and a topic name.";
  46
  47     /**
  48      * Instantiates MRClientBuilders.
  49      */
  50     private MRClientBuilders() {
  51         // prevent instantiation
  52     }
  53
  54     /**
  55      * A builder for a topic Consumer
  56      * @author author
  57      */
  58     public static class ConsumerBuilder
  59     {
  60
  61         /**
  62          * Set the host list
  63          * @param hostList a comma-separated list of hosts to use to connect to MR
  64          * @return this builder
  65          */
  66         public ConsumerBuilder usingHosts ( String hostList ) {
  67             return usingHosts ( MRConsumerImpl.stringToList(hostList) );
  68         }
  69
  70         /**
  71          * Set the host list
  72          * @param hostSet a set of hosts to use to connect to MR
  73          * @return this builder
  74          */
  75         public ConsumerBuilder usingHosts ( Collection<String> hostSet ) {
  76             fHosts = hostSet; return this;
  77         }
  78
  79         /**
  80          * Set the topic
  81          * @param topic the name of the topic to consume
  82          * @return this builder
  83          */
  84         public ConsumerBuilder onTopic ( String topic ) {
  85             fTopic=topic;
  86             return this;
  87         }
  88
  89         /**
  90          * Set the consumer's group and ID
  91          * @param consumerGroup The name of the consumer group this consumer is part of
  92          * @param consumerId The unique id of this consumer in its group
  93          * @return this builder
  94          */
  95         public ConsumerBuilder knownAs ( String consumerGroup, String consumerId ) {
  96             fGroup = consumerGroup;
  97             fId = consumerId;
  98             return this;
  99         }
 100
 101         /**
 102          * Set the API key and secret for this client.
 103          * @param apiKey
 104          * @param apiSecret
 105          * @return this builder
 106          */
 107         public ConsumerBuilder authenticatedBy ( String apiKey, String apiSecret ) {
 108             fApiKey = apiKey;
 109             fApiSecret = apiSecret;
 110             return this;
 111         }
 112
 113         /**
 114          * Set the server side timeout
 115          * @param timeoutMs    The amount of time in milliseconds that the server should keep the connection open while waiting for message traffic.
 116          * @return this builder
 117          */
 118         public ConsumerBuilder waitAtServer ( int timeoutMs ) {
 119             fTimeoutMs = timeoutMs;
 120             return this;
 121         };
 122
 123         /**
 124          * Set the maximum number of messages to receive per transaction
 125          * @param limit The maximum number of messages to receive from the server in one transaction.
 126          * @return this builder
 127          */
 128         public ConsumerBuilder receivingAtMost ( int limit ) {
 129             fLimit = limit;
 130             return this;
 131         };
 132
 133         /**
 134          * Set a filter to use on the server
 135          * @param filter a Highland Park standard library filter encoded in JSON
 136          * @return this builder
 137          */
 138         public ConsumerBuilder withServerSideFilter ( String filter ) {
 139             fFilter = filter;
 140             return this;
 141         }
 142
 143         /**
 144          * Build the consumer
 145          * @return a consumer
 146          */
 147         public MRConsumer build ()
 148         {
 149             if ( fHosts == null || fHosts.isEmpty() || fTopic == null )
 150             {
 151                 throw new IllegalArgumentException ( ILLEGAL_ARGUMENT_MESSAGE );
 152             }
 153
 154             if ( fGroup == null )
 155             {
 156                 fGroup = UUID.randomUUID ().toString ();
 157                 fId = "0";
 158                 log.info ( "Creating non-restartable client with group " + fGroup + " and ID " + fId + "." );
 159             }
 160
 161             if ( sfConsumerMock != null ) return sfConsumerMock;
 162             try {
 163                 return new MRConsumerImpl.MRConsumerImplBuilder().setHostPart(fHosts)
 164                         .setTopic(fTopic).setConsumerGroup(fGroup).setConsumerId(fId)
 165                         .setTimeoutMs(fTimeoutMs).setLimit(fLimit).setFilter(fFilter)
 166                         .setApiKey_username(fApiKey).setApiSecret_password(fApiSecret)
 167                         .createMRConsumerImpl();
 168             } catch (MalformedURLException e) {
 169                 throw new IllegalArgumentException(e);
 170             }
 171         }
 172
 173         private Collection<String> fHosts = null;
 174         private String fTopic = null;
 175         private String fGroup = null;
 176         private String fId = null;
 177         private String fApiKey = null;
 178         private String fApiSecret = null;
 179         private int fTimeoutMs = -1;
 180         private int fLimit = -1;
 181         private String fFilter = null;
 182     }
 183
 184     /*************************************************************************/
 185     /*************************************************************************/
 186     /*************************************************************************/
 187
 188     /**
 189      * A publisher builder
 190      * @author author
 191      */
 192     public static class PublisherBuilder
 193     {
 194
 195         /**
 196          * Set the MR/UEB host(s) to use
 197          * @param hostlist The host(s) used in the URL to MR. Can be "host:port", can be multiple comma-separated entries.
 198          * @return this builder
 199          */
 200         public PublisherBuilder usingHosts ( String hostlist ) {
 201             return usingHosts ( MRConsumerImpl.stringToList(hostlist) );
 202         }
 203
 204         /**
 205          * Set the MR/UEB host(s) to use
 206          * @param hostSet The host(s) used in the URL to MR. Can be "host:port"
 207          * @return this builder
 208          */
 209         public PublisherBuilder usingHosts ( String[] hostSet )
 210         {
 211             final TreeSet<String> hosts = new TreeSet<> ();
 212             for ( String hp : hostSet )
 213             {
 214                 hosts.add ( hp );
 215             }
 216             return usingHosts ( hosts );
 217         }
 218
 219         /**
 220          * Set the MR/UEB host(s) to use
 221          * @param hostlist The host(s) used in the URL to MR. Can be "host:port".
 222          * @return this builder
 223          */
 224         public PublisherBuilder usingHosts ( Collection<String> hostlist ) {
 225             fHosts=hostlist;
 226             return this;
 227         }
 228
 229         /**
 230          * Set the topic to publish on
 231          * @param topic The topic on which to publish messages.
 232          * @return this builder
 233          */
 234         public PublisherBuilder onTopic ( String topic ) {
 235             fTopic = topic;
 236             return this;
 237         }
 238
 239         /**
 240          * Batch message sends with the given limits.
 241          * @param messageCount The largest set of messages to batch.
 242          * @param ageInMs The maximum age of a message waiting in a batch.
 243          * @return this builder
 244          */
 245         public PublisherBuilder limitBatch ( int messageCount, int ageInMs ) {
 246             fMaxBatchSize = messageCount;
 247             fMaxBatchAgeMs = ageInMs;
 248             return this;
 249         }
 250
 251         /**
 252          * Compress transactions
 253          * @return this builder
 254          */
 255         public PublisherBuilder withCompresion () {
 256             return enableCompresion(true);
 257         }
 258
 259         /**
 260          * Do not compress transactions
 261          * @return this builder
 262          */
 263         public PublisherBuilder withoutCompresion () {
 264             return enableCompresion(false);
 265         }
 266
 267         /**
 268          * Set the compression option
 269          * @param compress true to gzip compress transactions
 270          * @return this builder
 271          */
 272         public PublisherBuilder enableCompresion ( boolean compress ) {
 273             fCompress = compress;
 274             return this;
 275         }
 276
 277         /**
 278          * Set the API key and secret for this client.
 279          * @param apiKey
 280          * @param apiSecret
 281          * @return this builder
 282          */
 283         public PublisherBuilder authenticatedBy ( String apiKey, String apiSecret ) {
 284             fApiKey = apiKey;
 285             fApiSecret = apiSecret;
 286             return this;
 287         }
 288
 289         /**
 290          * Build the publisher
 291          * @return a batching publisher
 292          */
 293         public MRBatchingPublisher build ()
 294         {
 295             if ( fHosts == null || fHosts.isEmpty() || fTopic == null )
 296             {
 297                 throw new IllegalArgumentException ( ILLEGAL_ARGUMENT_MESSAGE );
 298             }
 299
 300             if ( sfPublisherMock != null ) return sfPublisherMock;
 301
 302             final MRSimplerBatchPublisher pub = new MRSimplerBatchPublisher.Builder ().
 303                 againstUrls ( fHosts ).
 304                 onTopic ( fTopic ).
 305                 batchTo ( fMaxBatchSize, fMaxBatchAgeMs ).
 306                 compress ( fCompress ).
 307                 build ();
 308             if ( fApiKey != null )
 309             {
 310                 pub.setApiCredentials ( fApiKey, fApiSecret );
 311             }
 312             return pub;
 313         }
 314
 315         private Collection<String> fHosts = null;
 316         private String fTopic = null;
 317         private int fMaxBatchSize = 1;
 318         private int fMaxBatchAgeMs = 1;
 319         private boolean fCompress = false;
 320         private String fApiKey = null;
 321         private String fApiSecret = null;
 322     }
 323
 324     /**
 325      * A builder for an identity manager
 326      * @author author
 327      */
 328     public static class IdentityManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRIdentityManager>
 329     {
 330
 331         @Override
 332         protected MRIdentityManager constructClient ( Collection<String> hosts ) { try {
 333             return new MRMetaClient ( hosts );
 334         } catch (MalformedURLException e) {
 335             throw new IllegalArgumentException(e);
 336         } }
 337     }
 338
 339     /**
 340      * A builder for a topic manager
 341      * @author author
 342      */
 343     public static class TopicManagerBuilder extends AbstractAuthenticatedManagerBuilder<MRTopicManager>
 344     {
 345
 346         @Override
 347         protected MRTopicManager constructClient ( Collection<String> hosts ) { try {
 348             return new MRMetaClient ( hosts );
 349         } catch (MalformedURLException e) {
 350             throw new IllegalArgumentException(e);
 351         } }
 352     }
 353
 354     /**
 355      * Inject a consumer. Used to support unit tests.
 356      * @param cc
 357      */
 358     public static void $testInject ( MRConsumer cc )
 359     {
 360         sfConsumerMock = cc;
 361     }
 362
 363     /**
 364      * Inject a publisher. Used to support unit tests.
 365      * @param pub
 366      */
 367     public static void $testInject ( MRBatchingPublisher pub )
 368     {
 369         sfPublisherMock = pub;
 370     }
 371
 372     static MRConsumer sfConsumerMock = null;
 373     static MRBatchingPublisher sfPublisherMock = null;
 374
 375     /**
 376      * A builder for an identity manager
 377      * @author author
 378      */
 379     public static abstract class AbstractAuthenticatedManagerBuilder<T extends MRClient>
 380     {
 381         /**
 382          * Construct an identity manager builder.
 383          */
 384         public AbstractAuthenticatedManagerBuilder () {}
 385
 386         /**
 387          * Set the host list
 388          * @param hostList a comma-separated list of hosts to use to connect to MR
 389          * @return this builder
 390          */
 391         public AbstractAuthenticatedManagerBuilder<T> usingHosts ( String hostList ) {
 392             return usingHosts ( MRConsumerImpl.stringToList(hostList) );
 393         }
 394
 395         /**
 396          * Set the host list
 397          * @param hostSet a set of hosts to use to connect to MR
 398          * @return this builder
 399          */
 400         public AbstractAuthenticatedManagerBuilder<T> usingHosts ( Collection<String> hostSet ) {
 401             fHosts = hostSet;
 402             return this;
 403         }
 404
 405         /**
 406          * Set the API key and secret for this client.
 407          * @param apiKey
 408          * @param apiSecret
 409          * @return this builder
 410          */
 411         public AbstractAuthenticatedManagerBuilder<T> authenticatedBy ( String apiKey, String apiSecret ) {
 412             fApiKey = apiKey;
 413             fApiSecret = apiSecret;
 414             return this;
 415         }
 416
 417         /**
 418          * Build the consumer
 419          * @return a consumer
 420          */
 421         public T build ()
 422         {
 423              if ( fHosts.isEmpty() )
 424             {
 425                 throw new IllegalArgumentException ( ILLEGAL_ARGUMENT_MESSAGE );
 426             }
 427
 428             final T mgr = constructClient ( fHosts );
 429             mgr.setApiCredentials ( fApiKey, fApiSecret );
 430             return mgr;
 431         }
 432
 433         protected abstract T constructClient ( Collection<String> hosts );
 434
 435         private Collection<String> fHosts = null;
 436         private String fApiKey = null;
 437         private String fApiSecret = null;
 438     }
 439
 440     private static final Logger log = LoggerFactory.getLogger ( MRClientBuilders.class );
 441 }