Add key and timestamp to tailed messages (#35)

* Add key and timestmap to tailed messages

* Show key and timestamp together with message payload

* Cargo fmt the code

Author: sevenEng

resources/web_server/public/my_css.css

@@ -26,6 +26,14 @@ h3 {
     padding-bottom: 3px;
     margin-top: 0px;
     margin-bottom: 0px;
+    display: flex;
+}
+
+.message-key,
+.message-ts {
+    padding-right: 5px;
+    margin-right: 5px;
+    border-right: 1px solid #eee;
 }
 
 div.topic_tailer {

resources/web_server/public/my_js.js

@@ -106,6 +106,23 @@ function error_to_graphic(cell) {
     $(cell).html(symbol);
 }
 
+function message_to_tailer_entry(msg) {
+    var ts_text;
+    if (msg["created_at"]) {
+        ts_text = (new Date(msg["created_at"])).toISOString() + " Created";
+    } else if (msg["appended_at"]) {
+        ts_text = (new Date(msg["appended_at"])).toISOString() + " Appended";
+    } else {
+        ts_text = "N/A";
+    }
+
+    var entry = $("<div>", {class: "message"});
+    entry.append($("<div>", { class: "message-key", text: msg["key"] ? msg["key"] : "N/A" }));
+    entry.append($("<div>", { class: "message-ts", text: ts_text }));
+    entry.append($("<div>", { class: "message-payload", text: msg["payload"] }));
+    return entry;
+}
+
 $(document).ready(function() {
     $('#datatable-brokers-ajax').each(function(index) {
         $(this).DataTable({
@@ -375,9 +392,7 @@ function background_tailer(cluster_id, topic_name, tailer_id) {
       messages = JSON.parse(data);
       for (var i = 0; i < messages.length; i++) {
         var message = messages[i];
-        var p = $("<p>", {class: "message"});
-        p.append(truncate(message[2], max_msg_length));
-        div_tailer.append(p);
+        div_tailer.append(message_to_tailer_entry(message));
       }
       if (bottom)
           scroll_to_bottom(div_tailer);

src/live_consumer.rs

@@ -1,6 +1,7 @@
 use rdkafka::config::ClientConfig;
 use rdkafka::consumer::{BaseConsumer, Consumer, EmptyConsumerContext};
 use rdkafka::message::BorrowedMessage;
+use rdkafka::message::Timestamp::*;
 use rdkafka::Message;
 use rocket::http::RawStr;
 use rocket::State;
@@ -183,8 +184,18 @@ impl LiveConsumerStore {
 
 // TODO: check log in case of error
 
+#[derive(Serialize)]
+struct TailedMessage {
+    partition: i32,
+    offset: i64,
+    key: Option<String>,
+    created_at: Option<i64>,
+    appended_at: Option<i64>,
+    payload: String,
+}
+
 #[get("/api/tailer/<cluster_id>/<topic>/<id>")]
-pub fn test_live_consumer_api(
+pub fn topic_tailer_api(
     cluster_id: ClusterId,
     topic: &RawStr,
     id: u64,
@@ -217,16 +228,40 @@ pub fn test_live_consumer_api(
 
     let mut output = Vec::new();
     for message in consumer.poll(100, Duration::from_secs(3)) {
-        let payload = message
+        let key = message
+            .key()
+            .map(|bytes| String::from_utf8_lossy(bytes))
+            .map(|cow_str| cow_str.into_owned());
+
+        let mut created_at = None;
+        let mut appended_at = None;
+        match message.timestamp() {
+            CreateTime(ctime) => created_at = Some(ctime),
+            LogAppendTime(atime) => appended_at = Some(atime),
+            NotAvailable => (),
+        }
+
+        let original_payload = message
             .payload()
             .map(|bytes| String::from_utf8_lossy(bytes))
             .unwrap_or(Cow::Borrowed(""));
-        if payload.len() > 1024 {
-            let truncated = format!("{}...", payload.chars().take(1024).collect::<String>());
-            output.push(json! {(message.partition(), message.offset(), truncated)});
+        let payload = if original_payload.len() > 1024 {
+            format!(
+                "{}...",
+                original_payload.chars().take(1024).collect::<String>()
+            )
         } else {
-            output.push(json! {(message.partition(), message.offset(), payload)});
+            original_payload.into_owned()
         };
+
+        output.push(TailedMessage {
+            partition: message.partition(),
+            offset: message.offset(),
+            key,
+            created_at,
+            appended_at,
+            payload,
+        })
     }
 
     Ok(json!(output).to_string())

src/web_server/server.rs

@@ -127,7 +127,7 @@ pub fn run_server(executor: &ThreadPoolExecutor, cache: Cache, config: &Config)
                 api::topic_groups,
                 api::topic_search,
                 api::topic_topology,
-                live_consumer::test_live_consumer_api,
+                live_consumer::topic_tailer_api,
             ],
         )
         .launch();