Add timestamp() to librdkafka messages

Author: qix

confluent_kafka/src/confluent_kafka.c

@@ -164,7 +164,7 @@ static PyObject* KafkaError_richcompare (KafkaError *self, PyObject *o2,
 	Py_INCREF(result);
 	return result;
 }
-				       
+
 
 static PyTypeObject KafkaErrorType = {
 	PyVarObject_HEAD_INIT(NULL, 0)
@@ -326,6 +326,14 @@ static PyObject *Message_offset (Message *self, PyObject *ignore) {
 }
 
 
+static PyObject *Message_timestamp (Message *self, PyObject *ignore) {
+	if (self->timestamp > 0)
+		return PyLong_FromLong(self->timestamp);
+	else
+		Py_RETURN_NONE;
+}
+
+
 static PyMethodDef Message_methods[] = {
 	{ "error", (PyCFunction)Message_error, METH_NOARGS,
 	  "  The message object is also used to propagate errors and events, "
@@ -362,6 +370,11 @@ static PyMethodDef Message_methods[] = {
 	  "  :rtype: int or None\n"
 	  "\n"
 	},
+	{ "timestamp", (PyCFunction)Message_timestamp, METH_NOARGS,
+	  "  :returns: message timestamp or None if not available.\n"
+	  "  :rtype: int or None\n"
+	  "\n"
+	},
 	{ NULL }
 };
 
@@ -495,6 +508,12 @@ 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
+	}
+
 	return (PyObject *)self;
 }
 
@@ -560,7 +579,7 @@ static PyObject *TopicPartition_new (PyTypeObject *type, PyObject *args,
 
 	return TopicPartition_new0(topic, partition, offset, 0);
 }
-	
+
 
 
 static int TopicPartition_traverse (TopicPartition *self,
@@ -758,7 +777,7 @@ PyObject *c_parts_to_py (const rd_kafka_topic_partition_list_t *c_parts) {
 	}
 
 	return parts;
-		
+
 }
 
 /**

confluent_kafka/src/confluent_kafka.h

@@ -219,6 +219,7 @@ typedef struct {
 	PyObject *error;
 	int32_t partition;
 	int64_t offset;
+	int64_t timestamp;
 } Message;
 
 extern PyTypeObject MessageType;