kafka transport

.gitignore

@@ -0,0 +1,6 @@
+*~
+/composer.lock
+/composer.phar
+/phpunit.xml
+/vendor/
+/.idea/

.travis.yml

@@ -0,0 +1,21 @@
+sudo: false
+
+git:
+    depth: 1
+
+language: php
+
+php:
+    - '5.6'
+    - '7.0'
+
+cache:
+    directories:
+        - $HOME/.composer/cache
+
+install:
+    - composer self-update
+    - composer install --prefer-source --ignore-platform-reqs
+
+script:
+    - vendor/bin/phpunit --exclude-group=functional

LICENSE

@@ -0,0 +1,20 @@
+The MIT License (MIT)
+Copyright (c) 2016 Kotliar Maksym
+
+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,26 @@
+# RdKafka Transport
+
+[![Gitter](https://badges.gitter.im/php-enqueue/Lobby.svg)](https://gitter.im/php-enqueue/Lobby)
+[![Build Status](https://travis-ci.org/php-enqueue/rdkafka.png?branch=master)](https://travis-ci.org/php-enqueue/rdkafka)
+[![Total Downloads](https://poser.pugx.org/enqueue/rdkafka/d/total.png)](https://packagist.org/packages/enqueue/rdkafka)
+[![Latest Stable Version](https://poser.pugx.org/enqueue/rdkafka/version.png)](https://packagist.org/packages/enqueue/rdkafka)
+
+This is an implementation of PSR specification. It allows you to send and consume message via Kafka protocol.  
+
+## Resources
+
+* [Documentation](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/index.md)
+* [Questions](https://gitter.im/php-enqueue/Lobby)
+* [Issue Tracker](https://github.com/php-enqueue/enqueue-dev/issues)
+
+## Developed by Forma-Pro
+
+Forma-Pro is a full stack development company which interests also spread to open source development. 
+Being a team of strong professionals we have an aim an ability to help community by developing cutting edge solutions in the areas of e-commerce, docker & microservice oriented architecture where we have accumulated a huge many-years experience. 
+Our main specialization is Symfony framework based solution, but we are always looking to the technologies that allow us to do our job the best way. We are committed to creating solutions that revolutionize the way how things are developed in aspects of architecture & scalability.
+
+If you have any questions and inquires about our open source development, this product particularly or any other matter feel free to contact at opensource@forma-pro.com
+
+## License
+
+It is released under the [MIT License](LICENSE).

RdKafkaConnectionFactory.php

@@ -0,0 +1,80 @@
+<?php
+namespace Enqueue\RdKafka;
+
+use Interop\Queue\PsrConnectionFactory;
+
+class RdKafkaConnectionFactory implements PsrConnectionFactory
+{
+    /**
+     * @var array
+     */
+    private $config;
+
+    /**
+     * The config could be an array, string DSN or null. In case of null it will attempt to connect to localhost with default settings.
+     *
+     * [
+     *     'global' => [                                   // https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
+     *       'metadata.broker.list' => 'localhost:9092',
+     *     ],
+     *     'topic' => [],
+     *     'dr_msg_cb' => null,
+     *     'error_cb' => null,
+     *     'rebalance_cb' => null,
+     *     'partitioner' => null,                          // https://arnaud-lb.github.io/php-rdkafka/phpdoc/rdkafka-topicconf.setpartitioner.html
+     *     'log_level' => null,
+     * ]
+     *
+     * or
+     *
+     * rdkafka://host:port
+     *
+     * @param array|string $config
+     */
+    public function __construct($config = 'rdkafka://')
+    {
+        if (empty($config) || 'rdkafka://' === $config) {
+            $config = [];
+        } elseif (is_string($config)) {
+            $config = $this->parseDsn($config);
+        } elseif (is_array($config)) {
+        } else {
+            throw new \LogicException('The config must be either an array of options, a DSN string or null');
+        }
+
+        $this->config = array_replace($this->defaultConfig(), $config);
+    }
+
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return RdKafkaContext
+     */
+    public function createContext()
+    {
+        return new RdKafkaContext($this->config);
+    }
+
+    /**
+     * @param string $dsn
+     *
+     * @return array
+     */
+    private function parseDsn($dsn)
+    {
+
+    }
+
+    /**
+     * @return array
+     */
+    private function defaultConfig()
+    {
+        return [
+            'global' => [
+                'metadata.broker.list' => 'localhost:9092',
+            ],
+        ];
+    }
+}

RdKafkaContext.php

@@ -0,0 +1,150 @@
+<?php
+namespace Enqueue\RdKafka;
+
+use Interop\Queue\PsrConsumer;
+use Interop\Queue\PsrContext;
+use Interop\Queue\PsrDestination;
+use Interop\Queue\PsrMessage;
+use Interop\Queue\PsrProducer;
+use Interop\Queue\PsrQueue;
+use Interop\Queue\PsrTopic;
+use RdKafka\Conf;
+use RdKafka\Producer;
+use RdKafka\TopicConf;
+
+class RdKafkaContext implements PsrContext
+{
+    /**
+     * @var array
+     */
+    private $config;
+
+    /**
+     * @var Conf
+     */
+    private $conf;
+
+    /**
+     * @var Producer
+     */
+    private $producer;
+
+    /**
+     * @param array $config
+     */
+    public function __construct(array $config)
+    {
+        $this->config = $config;
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function createMessage($body = '', array $properties = [], array $headers = [])
+    {
+        return new RdKafkaMessage($body, $properties, $headers);
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return RdKafkaTopic
+     */
+    public function createTopic($topicName)
+    {
+        return new RdKafkaTopic($topicName);
+    }
+
+    public function createQueue($queueName)
+    {
+        // TODO: Implement createQueue() method.
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function createTemporaryQueue()
+    {
+        throw new \LogicException('Not implemented');
+    }
+
+    /**
+     * {@inheritdoc}
+     *
+     * @return RdKafkaProducer
+     */
+    public function createProducer()
+    {
+        return new RdKafkaProducer($this->getProducer());
+    }
+
+    public function createConsumer(PsrDestination $destination)
+    {
+        // TODO: Implement createConsumer() method.
+    }
+
+    public function close()
+    {
+
+    }
+
+    /**
+     * @return Producer
+     */
+    private function getProducer()
+    {
+        if (null === $this->producer) {
+            $this->producer = new Producer($this->getConf());
+
+            if (isset($this->config['log_level'])) {
+                $this->producer->setLogLevel($this->config['log_level']);
+            }
+        }
+
+        return $this->producer;
+    }
+
+    /**
+     * @return Conf
+     */
+    private function getConf()
+    {
+        if (null === $this->conf) {
+            $topicConf = new TopicConf();
+
+            if (isset($this->config['topic']) && is_array($this->config['topic'])) {
+                foreach ($this->config['topic'] as $key => $value) {
+                    $topicConf->set($key, $value);
+                }
+            }
+
+            if (isset($this->config['partitioner'])) {
+                $topicConf->setPartitioner($this->config['partitioner']);
+            }
+
+            $this->conf = new Conf();
+
+            if (isset($this->config['global']) && is_array($this->config['global'])) {
+                foreach ($this->config['global'] as $key => $value) {
+                    $this->conf->set($key, $value);
+                }
+            }
+
+            if (isset($this->config['dr_msg_cb'])) {
+                $this->conf->setDrMsgCb($this->config['dr_msg_cb']);
+            }
+
+            if (isset($this->config['error_cb'])) {
+                $this->conf->setErrorCb($this->config['error_cb']);
+            }
+
+            if (isset($this->config['rebalance_cb'])) {
+                $this->conf->setRebalanceCb($this->config['errorebalance_cbr_cb']);
+            }
+
+            $this->conf->setDefaultTopicConf($topicConf);
+        }
+
+        return $this->conf;
+    }
+}