"Fossies" - the Fresh Open Source Software Archive

Member "monasca-api-3.1.0/monasca_api/common/messaging/kafka_publisher.py" (27 Sep 2019, 2361 Bytes) of package /linux/misc/openstack/monasca-api-3.1.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. For more information about "kafka_publisher.py" see the Fossies "Dox" file reference documentation and the latest Fossies "Diffs" side-by-side code changes report: 3.0.0_vs_3.1.0.

    1 # Copyright 2014,2017 Hewlett-Packard
    2 #
    3 # Licensed under the Apache License, Version 2.0 (the "License"); you may
    4 # not use this file except in compliance with the License. You may obtain
    5 # a copy of the License at
    6 #
    7 #      http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 # Unless required by applicable law or agreed to in writing, software
   10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   12 # License for the specific language governing permissions and limitations
   13 # under the License.
   14 
   15 from monasca_common.kafka import client_factory
   16 import monasca_common.kafka_lib.common as kafka_common
   17 from oslo_config import cfg
   18 from oslo_log import log
   19 
   20 from monasca_api.common.messaging import exceptions
   21 from monasca_api.common.messaging import publisher
   22 
   23 
   24 LOG = log.getLogger(__name__)
   25 
   26 
   27 class KafkaPublisher(publisher.Publisher):
   28     def __init__(self, topic):
   29         if not cfg.CONF.kafka.uri:
   30             raise Exception('Kafka is not configured correctly! '
   31                             'Use configuration file to specify Kafka '
   32                             'uri, for example: '
   33                             'uri=192.168.1.191:9092')
   34 
   35         self.uri = cfg.CONF.kafka.uri
   36         self.topic = topic
   37         self.group = cfg.CONF.kafka.group
   38         self.wait_time = cfg.CONF.kafka.wait_time
   39         self.is_async = cfg.CONF.kafka.is_async
   40         self.ack_time = cfg.CONF.kafka.ack_time
   41         self.max_retry = cfg.CONF.kafka.max_retry
   42         self.auto_commit = cfg.CONF.kafka.auto_commit
   43         self.compact = cfg.CONF.kafka.compact
   44         self.partitions = cfg.CONF.kafka.partitions
   45         self.drop_data = cfg.CONF.kafka.drop_data
   46 
   47         self._producer = client_factory.get_kafka_producer(
   48             self.uri, cfg.CONF.kafka.legacy_kafka_client_enabled)
   49 
   50     def close(self):
   51         pass
   52 
   53     def send_message(self, message):
   54         try:
   55             self._producer.publish(self.topic, message)
   56 
   57         except (kafka_common.KafkaUnavailableError,
   58                 kafka_common.LeaderNotAvailableError):
   59             LOG.exception('Error occurred while posting data to Kafka.')
   60             raise exceptions.MessageQueueException()
   61         except Exception:
   62             LOG.exception('Unknown error.')
   63             raise exceptions.MessageQueueException()