This repository was archived by the owner on May 25, 2023. It is now read-only.

Commit b656167

author

Pavel Batanov

authored

Merge pull request #2 from lamoda/feature/rdkafka-4

Support RDKafka 4

2 parents 34215c6 + 005b531 commit b656167Copy full SHA for b656167

File tree

-1

lines changed

-1

lines changed

Lines changed: 2 additions & 1 deletion

Lines changed: 6 additions & 0 deletions

Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,7 @@`
`6`	`6`
`7`	`7`	`use Codeception\Module;`
`8`	`8`	`use Exception;`
	`9`	`+use Lamoda\Codeception\Extension\MessageSerializer\ArrayMessageSerializer;`
`9`	`10`	`use Lamoda\Codeception\Extension\MessageSerializer\MessageSerializerInterface;`
`10`	`11`	`use RdKafka\Conf;`
`11`	`12`	`use RdKafka\Consumer;`
`@@ -17,6 +18,7 @@`
`17`	`18`	`class KafkaModule extends Module`
`18`	`19`	`{`
`19`	`20`	`protected const DEFAULT_PARTITION = 0;`
	`21`	`+ protected const FLUSH_TIMEOUT_MS = 3000;`
`20`	`22`
`21`	`23`	`/**`
`22`	`24`	`* @var MessageSerializerInterface`
`@@ -83,6 +85,10 @@ public function putMessageInTopic(string $topicName, string $message, ?int $part`
`83`	`85`	`$topic = $producer->newTopic($topicName, $this->topicConf);`
`84`	`86`
`85`	`87`	`$topic->produce($partition ?? static::DEFAULT_PARTITION, 0, $message);`
	`88`	`+`
	`89`	`+ if (method_exists($producer, 'flush')) {`
	`90`	`+ $producer->flush(self::FLUSH_TIMEOUT_MS);`
	`91`	`+ }`
`86`	`92`	`}`
`87`	`93`
`88`	`94`	`public function putMessageListInTopic(string $topicName, array $messages, ?int $partition = null): void`

Comments

(0)