Updates for requested changes

Author: qix

confluent_kafka/src/confluent_kafka.c

@@ -327,13 +327,18 @@ static PyObject *Message_offset (Message *self, PyObject *ignore) {
 
 
 static PyObject *Message_timestamp (Message *self, PyObject *ignore) {
-	if (self->timestamp > 0)
+	if (self->tstype != RD_KAFKA_TIMESTAMP_NOT_AVAILABLE)
 		return PyLong_FromLong(self->timestamp);
 	else
 		Py_RETURN_NONE;
 }
 
 
+static PyObject *Message_tstype (Message *self, PyObject *ignore) {
+  return PyLong_FromLong(self->tstype);
+}
+
+
 static PyMethodDef Message_methods[] = {
 	{ "error", (PyCFunction)Message_error, METH_NOARGS,
 	  "  The message object is also used to propagate errors and events, "
@@ -375,6 +380,11 @@ static PyMethodDef Message_methods[] = {
 	  "  :rtype: int or None\n"
 	  "\n"
 	},
+	{ "tstype", (PyCFunction)Message_tstype, METH_NOARGS,
+	  "  :returns: message timestamp type.\n"
+	  "  :rtype: int\n"
+	  "\n"
+	},
 	{ NULL }
 };
 
@@ -508,11 +518,7 @@ PyObject *Message_new0 (const rd_kafka_message_t *rkm) {
 	self->partition = rkm->partition;
 	self->offset = rkm->offset;
 
-	rd_kafka_timestamp_type_t tstype;
-	self->timestamp = rd_kafka_message_timestamp(rkm, &tstype);
-	if (tstype != RD_KAFKA_TIMESTAMP_NOT_AVAILABLE) {
-		// todo: make tstype available to python api
-	}
+	self->timestamp = rd_kafka_message_timestamp(rkm, &self->tstype);
 
 	return (PyObject *)self;
 }

confluent_kafka/src/confluent_kafka.h

@@ -220,6 +220,7 @@ typedef struct {
 	int32_t partition;
 	int64_t offset;
 	int64_t timestamp;
+	rd_kafka_timestamp_type_t tstype;
 } Message;
 
 extern PyTypeObject MessageType;

examples/integration_test.py

@@ -210,6 +210,7 @@ def verify_consumer():
             'group.id': 'test.py',
             'session.timeout.ms': 6000,
             'enable.auto.commit': False,
+            'api.version.request': True,
             'on_commit': print_commit_result,
             'error_cb': error_cb,
             'default.topic.config': {
@@ -243,9 +244,9 @@ def verify_consumer():
                 break
 
         if False:
-            print('%s[%d]@%d: key=%s, value=%s' % \
+            print('%s[%d]@%d: key=%s, value=%s, tstype=%d, timestamp=%s' % \
                   (msg.topic(), msg.partition(), msg.offset(),
-                   msg.key(), msg.value()))
+                   msg.key(), msg.value(), msg.tstype(), repr(msg.timestamp())))
 
         if (msg.offset() % 5) == 0:
             # Async commit