Smalyshev has uploaded a new change for review.
https://gerrit.wikimedia.org/r/283750
Change subject: Create Kafka event relayer
......................................................................
Create Kafka event relayer
Bug: T125138
Change-Id: I9d7705cb164bc975c3a0ddf4a33ac54fe7de931c
---
M autoload.php
A includes/libs/eventrelayer/EventRelayerKafka.php
2 files changed, 66 insertions(+), 0 deletions(-)
git pull ssh://gerrit.wikimedia.org:29418/mediawiki/core
refs/changes/50/283750/1
diff --git a/autoload.php b/autoload.php
index c3e988c..d88db0e 100644
--- a/autoload.php
+++ b/autoload.php
@@ -397,6 +397,7 @@
'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',
'EventRelayerMCRD' => __DIR__ .
'/includes/libs/eventrelayer/EventRelayerMCRD.php',
'EventRelayerNull' => __DIR__ .
'/includes/libs/eventrelayer/EventRelayerNull.php',
'Exif' => __DIR__ . '/includes/media/Exif.php',
diff --git a/includes/libs/eventrelayer/EventRelayerKafka.php
b/includes/libs/eventrelayer/EventRelayerKafka.php
new file mode 100644
index 0000000..fc2bf23
--- /dev/null
+++ b/includes/libs/eventrelayer/EventRelayerKafka.php
@@ -0,0 +1,65 @@
+<?php
+use Kafka\Produce;
+
+/**
+ * Event relayer for Apache Kafka.
+ * Configuring for WANCache:
+ * 'relayerConfig' => [ 'class' => 'EventRelayerKafka',
+ * 'KafkaEventChannel' => 'wancache-purge',
'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 );
+ $this->channel = $this->config->get( 'KafkaEventChannel' );
+ if ( empty( $this->channel ) ) {
+ throw new InvalidArgumentException( "KafkaEventChannel
must be configured" );
+ }
+ 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 );
+ $producer = $this->getKafkaProducer();
+ $producer->setMessages( $this->channel, 0, $jsonEvents );
+ $producer->send();
+ }
+}
--
To view, visit https://gerrit.wikimedia.org/r/283750
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I9d7705cb164bc975c3a0ddf4a33ac54fe7de931c
Gerrit-PatchSet: 1
Gerrit-Project: mediawiki/core
Gerrit-Branch: master
Gerrit-Owner: Smalyshev <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
