track topics in message tracing

pb/trace.proto

@@ -39,6 +39,7 @@ message TraceEvent {
 
   message PublishMessage {
     optional bytes messageID = 1;
+    repeated string topics = 2;
   }
 
   message RejectMessage {
@@ -99,11 +100,16 @@ message TraceEvent {
   }
 
   message RPCMeta {
-    repeated bytes messageIDs = 1;
+    repeated MessageMeta messages = 1;
     repeated SubMeta subscription = 2;
     optional ControlMeta control = 3;
   }
 
+  message MessageMeta {
+    optional bytes messageID = 1;
+    repeated string topics = 2;
+  }
+
   message SubMeta {
     optional bool subscribe = 1;
     optional string topic = 2;

trace.go

@@ -31,6 +31,7 @@ func (t *pubsubTracer) PublishMessage(msg *Message) {
 		Timestamp: &now,
 		PublishMessage: &pb.TraceEvent_PublishMessage{
 			MessageID: []byte(msgID(msg.Message)),
+			Topics:    msg.Message.TopicIDs,
 		},
 	}
 
@@ -192,11 +193,14 @@ func (t *pubsubTracer) DropRPC(rpc *RPC, p peer.ID) {
 func traceRPCMeta(rpc *RPC) *pb.TraceEvent_RPCMeta {
 	rpcMeta := new(pb.TraceEvent_RPCMeta)
 
-	var mids [][]byte
+	var msgs []*pb.TraceEvent_MessageMeta
 	for _, m := range rpc.Publish {
-		mids = append(mids, []byte(msgID(m)))
+		msgs = append(msgs, &pb.TraceEvent_MessageMeta{
+			MessageID: []byte(msgID(m)),
+			Topics:    m.TopicIDs,
+		})
 	}
-	rpcMeta.MessageIDs = mids
+	rpcMeta.Messages = msgs
 
 	var subs []*pb.TraceEvent_SubMeta
 	for _, sub := range rpc.Subscriptions {