gerrit.onap Code Review - policy/common.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * Copyright (C) 2022 Nordix Foundation.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  * ============LICENSE_END=========================================================
  17  */
  18
  19 package org.onap.policy.common.endpoints.event.comm.bus.internal;
  20
  21 import org.onap.policy.common.endpoints.event.comm.Topic;
  22 import org.onap.policy.common.endpoints.event.comm.bus.KafkaTopicSink;
  23 import org.slf4j.Logger;
  24 import org.slf4j.LoggerFactory;
  25
  26 /**
  27  * This implementation publishes events for the associated KAFKA topic, inline with the calling
  28  * thread.
  29  */
  30 public class InlineKafkaTopicSink extends InlineBusTopicSink implements KafkaTopicSink {
  31
  32     /**
  33      * Logger.
  34      */
  35     private static Logger logger = LoggerFactory.getLogger(InlineKafkaTopicSink.class);
  36
  37     /**
  38      * Argument-based KAFKA Topic Writer instantiation. BusTopicParams contains below mentioned
  39      * attributes.
  40      *
  41      * <p>servers              list of KAFKA servers available for publishing
  42      * topic                the topic to publish to
  43      * partitionId          the partition key (optional, autogenerated if not provided)
  44      * useHttps             does connection use HTTPS?
  45      * @param busTopicParams contains attributes needed
  46      * @throws IllegalArgumentException if invalid arguments are detected
  47      */
  48     public InlineKafkaTopicSink(BusTopicParams busTopicParams) {
  49         super(busTopicParams);
  50     }
  51
  52     /**
  53      * Instantiation of internal resources.
  54      */
  55     @Override
  56     public void init() {
  57
  58         this.publisher = new BusPublisher.KafkaPublisherWrapper(BusTopicParams.builder()
  59                 .servers(this.servers)
  60                 .topic(this.effectiveTopic)
  61                 .useHttps(this.useHttps)
  62                 .build());
  63         logger.info("{}: KAFKA SINK created", this);
  64     }
  65
  66     @Override
  67     public String toString() {
  68         return "InlineKafkaTopicSink [getTopicCommInfrastructure()=" + getTopicCommInfrastructure() + ", toString()="
  69                         + super.toString() + "]";
  70     }
  71
  72     @Override
  73     public CommInfrastructure getTopicCommInfrastructure() {
  74         return Topic.CommInfrastructure.KAFKA;
  75     }
  76 }