Skip to content

Commit c7bb45f

Browse files
ASKozienkoASKozienko
committed
kafka transport
1 parent 6d61c3c commit c7bb45f

17 files changed

+843
-1
lines changed

‎composer.json‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,8 @@
1414
"enqueue/sqs": "*@dev",
1515
"enqueue/pheanstalk": "*@dev",
1616
"enqueue/gearman": "*@dev",
17+
"enqueue/rdkafka": "*@dev",
18+
"kwn/php-rdkafka-stubs": "^1.0.2",
1719
"enqueue/enqueue-bundle": "*@dev",
1820
"enqueue/job-queue": "*@dev",
1921
"enqueue/simple-client": "*@dev",
@@ -93,6 +95,10 @@
9395
"type": "path",
9496
"url": "pkg/gearman"
9597
},
98+
{
99+
"type": "path",
100+
"url": "pkg/rdkafka"
101+
},
96102
{
97103
"type": "path",
98104
"url": "pkg/simple-client"

‎docker-compose.yml‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,20 @@ services:
7171
volumes:
7272
- ./:/mqdev
7373

74+
zookeeper:
75+
image: 'wurstmeister/zookeeper'
76+
ports:
77+
- '2181:2181'
78+
79+
kafka:
80+
image: 'wurstmeister/kafka:0.10.2.1'
81+
ports:
82+
- '9092:9092'
83+
environment:
84+
KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
85+
volumes:
86+
- '/var/run/docker.sock:/var/run/docker.sock'
87+
7488
volumes:
7589
mysql-data:
7690
driver: local

‎docker/Dockerfile‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,19 @@ FROM formapro/nginx-php-fpm:latest-all-exts
33
## libs
44
RUN set -x && \
55
apt-get update && \
6-
apt-get install -y --no-install-recommends wget curl openssl ca-certificates nano netcat php-redis
6+
apt-get install -y wget curl openssl ca-certificates nano netcat php-dev php-redis git python
77

88
## confis
99

1010
# RUN rm -f /etc/php/7.0/cli/conf.d/*xdebug.ini
1111

12+
## librdkafka
13+
RUN git clone https://github.com/edenhill/librdkafka.git /root/librdkafka
14+
RUN cd /root/librdkafka && git checkout v0.11.0-RC2 && ./configure && make && make install
15+
RUN pecl install rdkafka
16+
RUN echo "extension=rdkafka.so" > /etc/php/7.1/cli/conf.d/10-rdkafka.ini
17+
RUN echo "extension=rdkafka.so" > /etc/php/7.1/fpm/conf.d/10-rdkafka.ini
18+
1219
COPY ./php/cli.ini /etc/php/7.1/cli/conf.d/1-dev_cli.ini
1320
COPY ./bin/dev_entrypoiny.sh /usr/local/bin/entrypoint.sh
1421
RUN chmod u+x /usr/local/bin/entrypoint.sh

‎phpunit.xml.dist‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,10 @@
5353
<directory>pkg/gearman/Tests</directory>
5454
</testsuite>
5555

56+
<testsuite name="rdkafka transport">
57+
<directory>pkg/rdkafka/Tests</directory>
58+
</testsuite>
59+
5660
<testsuite name="enqueue-bundle">
5761
<directory>pkg/enqueue-bundle/Tests</directory>
5862
</testsuite>

‎pkg/rdkafka/.gitignore‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
*~
2+
/composer.lock
3+
/composer.phar
4+
/phpunit.xml
5+
/vendor/
6+
/.idea/

‎pkg/rdkafka/.travis.yml‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
sudo: false
2+
3+
git:
4+
depth: 1
5+
6+
language: php
7+
8+
php:
9+
- '5.6'
10+
- '7.0'
11+
12+
cache:
13+
directories:
14+
- $HOME/.composer/cache
15+
16+
install:
17+
- composer self-update
18+
- composer install --prefer-source --ignore-platform-reqs
19+
20+
script:
21+
- vendor/bin/phpunit --exclude-group=functional

‎pkg/rdkafka/LICENSE‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
The MIT License (MIT)
2+
Copyright (c) 2016 Kotliar Maksym
3+
4+
Permission is hereby granted, free of charge, to any person obtaining a copy
5+
of this software and associated documentation files (the "Software"), to deal
6+
in the Software without restriction, including without limitation the rights
7+
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
8+
copies of the Software, and to permit persons to whom the Software is furnished
9+
to do so, subject to the following conditions:
10+
11+
The above copyright notice and this permission notice shall be included in all
12+
copies or substantial portions of the Software.
13+
14+
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
15+
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
16+
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
17+
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
18+
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
19+
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
20+
THE SOFTWARE.

‎pkg/rdkafka/README.md‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
# RdKafka Transport
2+
3+
[![Gitter](https://badges.gitter.im/php-enqueue/Lobby.svg)](https://gitter.im/php-enqueue/Lobby)
4+
[![Build Status](https://travis-ci.org/php-enqueue/rdkafka.png?branch=master)](https://travis-ci.org/php-enqueue/rdkafka)
5+
[![Total Downloads](https://poser.pugx.org/enqueue/rdkafka/d/total.png)](https://packagist.org/packages/enqueue/rdkafka)
6+
[![Latest Stable Version](https://poser.pugx.org/enqueue/rdkafka/version.png)](https://packagist.org/packages/enqueue/rdkafka)
7+
8+
This is an implementation of PSR specification. It allows you to send and consume message via Kafka protocol.
9+
10+
## Resources
11+
12+
* [Documentation](https://github.com/php-enqueue/enqueue-dev/blob/master/docs/index.md)
13+
* [Questions](https://gitter.im/php-enqueue/Lobby)
14+
* [Issue Tracker](https://github.com/php-enqueue/enqueue-dev/issues)
15+
16+
## Developed by Forma-Pro
17+
18+
Forma-Pro is a full stack development company which interests also spread to open source development.
19+
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.
20+
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.
21+
22+
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
23+
24+
## License
25+
26+
It is released under the [MIT License](LICENSE).

‎pkg/rdkafka/RdKafkaConnectionFactory.php‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
<?php
2+
namespace Enqueue\RdKafka;
3+
4+
use Interop\Queue\PsrConnectionFactory;
5+
6+
class RdKafkaConnectionFactory implements PsrConnectionFactory
7+
{
8+
/**
9+
* @var array
10+
*/
11+
private $config;
12+
13+
/**
14+
* The config could be an array, string DSN or null. In case of null it will attempt to connect to localhost with default settings.
15+
*
16+
* [
17+
* 'global' => [ // https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
18+
* 'metadata.broker.list' => 'localhost:9092',
19+
* ],
20+
* 'topic' => [],
21+
* 'dr_msg_cb' => null,
22+
* 'error_cb' => null,
23+
* 'rebalance_cb' => null,
24+
* 'partitioner' => null, // https://arnaud-lb.github.io/php-rdkafka/phpdoc/rdkafka-topicconf.setpartitioner.html
25+
* 'log_level' => null,
26+
* ]
27+
*
28+
* or
29+
*
30+
* rdkafka://host:port
31+
*
32+
* @param array|string $config
33+
*/
34+
public function __construct($config = 'rdkafka://')
35+
{
36+
if (empty($config) || 'rdkafka://' === $config) {
37+
$config = [];
38+
} elseif (is_string($config)) {
39+
$config = $this->parseDsn($config);
40+
} elseif (is_array($config)) {
41+
} else {
42+
throw new \LogicException('The config must be either an array of options, a DSN string or null');
43+
}
44+
45+
$this->config = array_replace($this->defaultConfig(), $config);
46+
}
47+
48+
49+
/**
50+
* {@inheritdoc}
51+
*
52+
* @return RdKafkaContext
53+
*/
54+
public function createContext()
55+
{
56+
return new RdKafkaContext($this->config);
57+
}
58+
59+
/**
60+
* @param string $dsn
61+
*
62+
* @return array
63+
*/
64+
private function parseDsn($dsn)
65+
{
66+
67+
}
68+
69+
/**
70+
* @return array
71+
*/
72+
private function defaultConfig()
73+
{
74+
return [
75+
'global' => [
76+
'metadata.broker.list' => 'localhost:9092',
77+
],
78+
];
79+
}
80+
}

‎pkg/rdkafka/RdKafkaContext.php‎

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
<?php
2+
namespace Enqueue\RdKafka;
3+
4+
use Interop\Queue\PsrConsumer;
5+
use Interop\Queue\PsrContext;
6+
use Interop\Queue\PsrDestination;
7+
use Interop\Queue\PsrMessage;
8+
use Interop\Queue\PsrProducer;
9+
use Interop\Queue\PsrQueue;
10+
use Interop\Queue\PsrTopic;
11+
use RdKafka\Conf;
12+
use RdKafka\Producer;
13+
use RdKafka\TopicConf;
14+
15+
class RdKafkaContext implements PsrContext
16+
{
17+
/**
18+
* @var array
19+
*/
20+
private $config;
21+
22+
/**
23+
* @var Conf
24+
*/
25+
private $conf;
26+
27+
/**
28+
* @var Producer
29+
*/
30+
private $producer;
31+
32+
/**
33+
* @param array $config
34+
*/
35+
public function __construct(array $config)
36+
{
37+
$this->config = $config;
38+
}
39+
40+
/**
41+
* {@inheritdoc}
42+
*/
43+
public function createMessage($body = '', array $properties = [], array $headers = [])
44+
{
45+
return new RdKafkaMessage($body, $properties, $headers);
46+
}
47+
48+
/**
49+
* {@inheritdoc}
50+
*
51+
* @return RdKafkaTopic
52+
*/
53+
public function createTopic($topicName)
54+
{
55+
return new RdKafkaTopic($topicName);
56+
}
57+
58+
public function createQueue($queueName)
59+
{
60+
// TODO: Implement createQueue() method.
61+
}
62+
63+
/**
64+
* {@inheritdoc}
65+
*/
66+
public function createTemporaryQueue()
67+
{
68+
throw new \LogicException('Not implemented');
69+
}
70+
71+
/**
72+
* {@inheritdoc}
73+
*
74+
* @return RdKafkaProducer
75+
*/
76+
public function createProducer()
77+
{
78+
return new RdKafkaProducer($this->getProducer());
79+
}
80+
81+
public function createConsumer(PsrDestination $destination)
82+
{
83+
// TODO: Implement createConsumer() method.
84+
}
85+
86+
public function close()
87+
{
88+
89+
}
90+
91+
/**
92+
* @return Producer
93+
*/
94+
private function getProducer()
95+
{
96+
if (null === $this->producer) {
97+
$this->producer = new Producer($this->getConf());
98+
99+
if (isset($this->config['log_level'])) {
100+
$this->producer->setLogLevel($this->config['log_level']);
101+
}
102+
}
103+
104+
return $this->producer;
105+
}
106+
107+
/**
108+
* @return Conf
109+
*/
110+
private function getConf()
111+
{
112+
if (null === $this->conf) {
113+
$topicConf = new TopicConf();
114+
115+
if (isset($this->config['topic']) && is_array($this->config['topic'])) {
116+
foreach ($this->config['topic'] as $key => $value) {
117+
$topicConf->set($key, $value);
118+
}
119+
}
120+
121+
if (isset($this->config['partitioner'])) {
122+
$topicConf->setPartitioner($this->config['partitioner']);
123+
}
124+
125+
$this->conf = new Conf();
126+
127+
if (isset($this->config['global']) && is_array($this->config['global'])) {
128+
foreach ($this->config['global'] as $key => $value) {
129+
$this->conf->set($key, $value);
130+
}
131+
}
132+
133+
if (isset($this->config['dr_msg_cb'])) {
134+
$this->conf->setDrMsgCb($this->config['dr_msg_cb']);
135+
}
136+
137+
if (isset($this->config['error_cb'])) {
138+
$this->conf->setErrorCb($this->config['error_cb']);
139+
}
140+
141+
if (isset($this->config['rebalance_cb'])) {
142+
$this->conf->setRebalanceCb($this->config['errorebalance_cbr_cb']);
143+
}
144+
145+
$this->conf->setDefaultTopicConf($topicConf);
146+
}
147+
148+
return $this->conf;
149+
}
150+
}

0 commit comments

Comments
 (0)