'ErrorPageError' => __DIR__ . '/includes/exception/ErrorPageError.php',
'EventRelayer' => __DIR__ . '/includes/libs/eventrelayer/EventRelayer.php',
'EventRelayerGroup' => __DIR__ . '/includes/EventRelayerGroup.php',
+ 'EventRelayerKafka' => __DIR__ . '/includes/libs/eventrelayer/EventRelayerKafka.php',
'EventRelayerNull' => __DIR__ . '/includes/libs/eventrelayer/EventRelayerNull.php',
'Exif' => __DIR__ . '/includes/media/Exif.php',
'ExifBitmapHandler' => __DIR__ . '/includes/media/ExifBitmap.php',
* @file
* @author Aaron Schulz
*/
+use Psr\Log\LoggerInterface;
+use Psr\Log\LoggerAwareInterface;
+use Psr\Log\NullLogger;
/**
* Base class for reliable event relays
*/
-abstract class EventRelayer {
+abstract class EventRelayer implements LoggerAwareInterface {
+ /** @var LoggerInterface */
+ protected $logger;
+
/**
* @param array $params
*/
public function __construct( array $params ) {
+ $this->logger = new NullLogger();
}
/**
return $this->doNotify( $channel, $events );
}
+ /**
+ * Set logger instance.
+ * @param LoggerInterface $logger
+ */
+ public function setLogger( LoggerInterface $logger ) {
+ $this->logger = $logger;
+ }
+
/**
* @param string $channel
* @param array $events List of event data maps
--- /dev/null
+<?php
+use Kafka\Produce;
+
+/**
+ * Event relayer for Apache Kafka.
+ * Configuring for WANCache:
+ * 'relayerConfig' => [ 'class' => 'EventRelayerKafka', 'KafkaEventHost' => 'localhost:9092' ],
+ */
+class EventRelayerKafka extends EventRelayer {
+
+ /**
+ * Configuration.
+ *
+ * @var Config
+ */
+ protected $config;
+
+ /**
+ * Kafka producer.
+ *
+ * @var Produce
+ */
+ protected $producer;
+
+ /**
+ * Create Kafka producer.
+ *
+ * @param Config $config
+ */
+ public function __construct( array $params ) {
+ $this->config = new HashConfig( $params );
+ if ( !$this->config->has( 'KafkaEventHost' ) ) {
+ throw new InvalidArgumentException( "KafkaEventHost must be configured" );
+ }
+ }
+
+ /**
+ * Get the producer object from kafka-php.
+ * @return Produce
+ */
+ protected function getKafkaProducer() {
+ if ( !$this->producer ) {
+ $this->producer = Produce::getInstance( null, null, $this->config->get( 'KafkaEventHost' ) );
+ }
+ return $this->producer;
+ }
+
+ /**
+ * (non-PHPdoc)
+ *
+ * @see EventRelayer::doNotify()
+ *
+ */
+ 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;
+ }
+}