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