First commit

CHANGELOG.md

@@ -0,0 +1,7 @@
+# Release Notes
+
+## [Unreleased](https://github.com/SepteniTechnology/laravel-kafka/compare/v1.0.0...master)
+
+## [v1.0.0 - 2020-09-19](https://github.com/SepteniTechnology/laravel-kafka/tree/v1.0.0)
+### Added
+- First launch 🚀🚀🚀

LICENSE.md

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) Septeni Technology
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

README.md

@@ -0,0 +1,43 @@
+
+## Installation
+
+```shell script
+composer require septech-laravel/kafka 
+```
+
+## Usage
+
+### Via injection
+
+```php
+<?php
+
+use Septech\Kafka\Kafka;
+
+class MessageController
+{
+    public function send(Kafka $kafka)
+    {
+        $message = json_encode(['message' => 'Hi, from kafka with love <3']);
+        $kafka->produce("some.topic", $message);
+    }
+}
+
+```
+
+### Via Facade
+
+```php
+<?php
+
+use Septech\Kafka\Facades\Kafka;
+
+$message = json_encode(['message' => 'Hi, from kafka with love <3']);
+Kafka::produce("some.topic", $message);
+```
+
+## Look like package is missing something?
+
+Well, nothing this absolutely perfect at first time.
+I just create this package for my purpose. If you want more features for this package.
+It needs you. Feel free send me a PR.

composer.json

@@ -0,0 +1,30 @@
+{
+  "name": "septech-laravel/kafka",
+  "description": "Rdkafka connector for Laravel",
+  "license": ["MIT"],
+  "authors": [
+    {
+      "name": "Đỗ Anh Tuấn",
+      "email": "tuan_da@septeni-technology.jp"
+    }
+  ],
+  "autoload": {
+    "psr-4": {
+      "Septech\\Kafka\\": "src"
+    }
+  },
+  "require": {
+    "ext-rdkafka": "*",
+    "illuminate/support": "^v7.0.0"
+  },
+  "config": {
+    "sort-packages": true
+  },
+  "extra": {
+    "laravel": {
+      "providers": [
+        "Septech\\Kafka\\KafkaServerProvider"
+      ]
+    }
+  }
+}

config/kafka.php

@@ -0,0 +1,8 @@
+<?php
+
+return [
+    // Brokers address
+    'brokers' => [
+        "10.0.0.70:9092"
+    ],
+];

src/Exceptions/FlushingException.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace Septech\Kafka\Exceptions;
+
+class FlushingException extends KafkaException
+{
+}

src/Exceptions/KafkaException.php

@@ -0,0 +1,10 @@
+<?php
+
+namespace Septech\Kafka\Exceptions;
+
+use Exception;
+
+class KafkaException extends Exception
+{
+
+}

src/Facades/Kafka.php

@@ -0,0 +1,19 @@
+<?php
+
+namespace Septech\Kafka\Facades;
+
+use RdKafka\Topic;
+use Illuminate\Support\Facades\Facade;
+
+/**
+ * @method static Topic topic(string $name)
+ * @method static \Septech\Kafka\Kafka addBroker(array $brokers)
+ * @method static int produce(string $name, mixed $payload, $timeout_ms = 10000)
+ */
+class Kafka extends Facade
+{
+    protected static function getFacadeAccessor()
+    {
+        return \Septech\Kafka\Kafka::class;
+    }
+}

src/Kafka.php

@@ -0,0 +1,63 @@
+<?php
+
+namespace Septech\Kafka;
+
+use RdKafka;
+use Septech\Kafka\Exceptions\FlushingException;
+
+/**
+ * @mixin RdKafka
+ */
+class Kafka
+{
+    /**
+     * @var RdKafka
+     */
+    protected $kafka;
+
+    public function __construct(RdKafka $kafka, array $brokers = [])
+    {
+        $this->kafka = $kafka;
+
+        $this->addBroker($brokers);
+    }
+
+    public function topic(string $name)
+    {
+        return $this->kafka->newTopic($name);
+    }
+
+    public function addBroker(array $brokers)
+    {
+        $this->kafka->addBrokers(implode(',', $brokers));
+
+        return $this;
+    }
+
+    /**
+     * @param string $topic
+     * @param $payload
+     * @param int $timeout_ms
+     * @return mixed
+     * @throws FlushingException
+     */
+    public function produce(string $topic, $payload, $timeout_ms = 10000)
+    {
+        $this->topic($topic)->produce(RD_KAFKA_PARTITION_UA, 0, $payload);
+
+        $result = $this->kafka->flush($timeout_ms);
+
+        if ($result != RD_KAFKA_RESP_ERR_NO_ERROR) {
+            throw new FlushingException(
+                "An error occur while flushing messages to kafka. Messages might be lost"
+            );
+        }
+
+        return $result;
+    }
+
+    public function __call($method, $arguments)
+    {
+        $this->kafka->$method(...$arguments);
+    }
+}

src/KafkaServerProvider.php

@@ -0,0 +1,37 @@
+<?php
+
+namespace Septech\Kafka;
+
+use RdKafka;
+use RdKafka\Conf;
+use RdKafka\Producer;
+use Illuminate\Support\ServiceProvider;
+
+class KafkaServerProvider extends ServiceProvider
+{
+    public function register()
+    {
+        $this->app->singleton(Kafka::class, function ($app) {
+            $config = $app['config']->get('kafka');
+
+            return new Kafka(
+                $app->make(RdKafka::class),
+                $brokers = $config['brokers'] ?? []
+            );
+        });
+
+
+        $this->app->singleton(RdKafka::class, function () {
+            $conf = new Conf();
+            $conf->set('log_level', (string) LOG_DEBUG);
+            $conf->set('debug', 'all');
+            return new Producer($conf);
+        });
+
+        $this->mergeConfigFrom(__DIR__ . '/../config/kafka.php', 'kafka');
+
+        $this->publishes([
+            __DIR__ . '/../config/kafka.php' => config_path('kafka.php')
+        ], 'kafka');
+    }
+}

src/Message.php

@@ -0,0 +1,21 @@
+<?php
+
+namespace Septech\Kafka;
+
+abstract class Message
+{
+    public function __toString()
+    {
+        return $this->toJson();
+    }
+
+    public function toJson($options = 0)
+    {
+        return json_encode($this->toArray(), $options);
+    }
+
+    public function toArray()
+    {
+        return get_object_vars($this);
+    }
+}