EventRelayerKafka.php

Go to the documentation of this file.

<?php
 
use Kafka\Produce;
 
class EventRelayerKafka extends EventRelayer {
    protected $config;
 
    protected $producer;
 
    public function __construct( array $params ) {
        parent::__construct( $params );
 
        $this->config = new HashConfig( $params );
        if ( !$this->config->has( 'KafkaEventHost' ) ) {
            throw new InvalidArgumentException( "KafkaEventHost must be configured" );
        }
    }
 
    protected function getKafkaProducer() {
        if ( !$this->producer ) {
            $this->producer = Produce::getInstance(
                null, null, $this->config->get( 'KafkaEventHost' ) );
        }
        return $this->producer;
    }
 
    protected function doNotify( $channel, array $events ) {
        $jsonEvents = array_map( 'json_encode', $events );
        try {
            $producer = $this->getKafkaProducer();
            $producer->setMessages( $channel, 0, $jsonEvents );
            $producer->send();
        } catch ( \Kafka\Exception $e ) {
            $this->logger->warning( "Sending events failed: $e" );
            return false;
        }
        return true;
    }
}