Merge pull request arnaud-lb#179 from awons/awons/allow-message-deletion

Allow for null key and null message

Author: arnaud-lb

tests/allow_null_payload.phpt

@@ -0,0 +1,46 @@
+--TEST--
+Allow null payload
+--SKIPIF--
+<?php
+file_exists(__DIR__."/test_env.php") || die("skip");
+--FILE--
+<?php
+
+require __DIR__."/test_env.php";
+
+$topicName = sprintf('test_rdkafka_%s', uniqid());
+
+$producer = new RdKafka\Producer();
+$producer->addBrokers(TEST_KAFKA_BROKERS);
+$topic = $producer->newTopic($topicName);
+
+$topic->produce(0, 0, NULL, 'message_key_1');
+
+while ($producer->getOutQLen() > 0) {
+ $producer->poll(50);
+}
+
+$consumer = new RdKafka\Consumer();
+$consumer->addBrokers(TEST_KAFKA_BROKERS);
+
+$topic = $consumer->newTopic($topicName);
+$topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);
+
+while (true) {
+ $message = $topic->consume(0, 1000);
+ if ($message === null) {
+ continue;
+ }
+ switch ($message->err) {
+ case RD_KAFKA_RESP_ERR_NO_ERROR:
+ var_dump($message->payload);
+ var_dump($message->key);
+ break 2;
+ }
+}
+
+$topic->consumeStop(0);
+
+--EXPECTF--
+NULL
+string(13) "message_key_1"

tests/allow_null_payload_and_key.phpt

@@ -0,0 +1,46 @@
+--TEST--
+Allow null payload
+--SKIPIF--
+<?php
+file_exists(__DIR__."/test_env.php") || die("skip");
+--FILE--
+<?php
+
+require __DIR__."/test_env.php";
+
+$topicName = sprintf('test_rdkafka_%s', uniqid());
+
+$producer = new RdKafka\Producer();
+$producer->addBrokers(TEST_KAFKA_BROKERS);
+$topic = $producer->newTopic($topicName);
+
+$topic->produce(0, 0);
+
+while ($producer->getOutQLen() > 0) {
+ $producer->poll(50);
+}
+
+$consumer = new RdKafka\Consumer();
+$consumer->addBrokers(TEST_KAFKA_BROKERS);
+
+$topic = $consumer->newTopic($topicName);
+$topic->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);
+
+while (true) {
+ $message = $topic->consume(0, 1000);
+ if ($message === null) {
+ continue;
+ }
+ switch ($message->err) {
+ case RD_KAFKA_RESP_ERR_NO_ERROR:
+ var_dump($message->payload);
+ var_dump($message->key);
+ break 2;
+ }
+}
+
+$topic->consumeStop(0);
+
+--EXPECTF--
+NULL
+NULL

topic.c

@@ -353,10 +353,10 @@ static const zend_function_entry kafka_kafka_consumer_topic_fe[] = {
  PHP_FE_END
 };
 
-/* {{{ proto void RdKafka\ProducerTopic::produce(int $partition, int $msgflags, string $payload[, string $key])
+/* {{{ proto void RdKafka\ProducerTopic::produce(int $partition, int $msgflags[, string $payload, string $key])
  Produce and send a single message to broker. */
 
-ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_produce, 0, 0, 3)
+ZEND_BEGIN_ARG_INFO_EX(arginfo_kafka_produce, 0, 0, 2)
  ZEND_ARG_INFO(0, partition)
  ZEND_ARG_INFO(0, msgflags)
  ZEND_ARG_INFO(0, payload)
@@ -367,15 +367,15 @@ PHP_METHOD(RdKafka__ProducerTopic, produce)
 {
  long partition;
  long msgflags;
- char *payload;
- arglen_t payload_len;
+ char *payload = NULL;
+ arglen_t payload_len = 0;
  char *key = NULL;
  arglen_t key_len = 0;
  int ret;
  rd_kafka_resp_err_t err;
  kafka_topic_object *intern;
 
- if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "lls|s!", &partition, &msgflags, &payload, &payload_len, &key, &key_len) == FAILURE) {
+ if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "ll|s!s!", &partition, &msgflags, &payload, &payload_len, &key, &key_len) == FAILURE) {
  return;
  }