add sender id to message

go.mod

@@ -11,3 +11,5 @@ require (
 	github.com/multiformats/go-multistream v0.1.0
 	github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee
 )
+
+go 1.13

pubsub.go

@@ -135,12 +135,17 @@ type PubSubRouter interface {
 
 type Message struct {
 	*pb.Message
+	senderID peer.ID
 }
 
 func (m *Message) GetFrom() peer.ID {
 	return peer.ID(m.Message.GetFrom())
 }
 
+func (m *Message) GetSenderID() peer.ID {
+	return m.senderID
+}
+
 type RPC struct {
 	pb.RPC
 
@@ -522,12 +527,15 @@ func (p *PubSub) doAnnounceRetry(pid peer.ID, topic string, sub bool) {
 
 // notifySubs sends a given message to all corresponding subscribers.
 // Only called from processLoop.
-func (p *PubSub) notifySubs(msg *pb.Message) {
+func (p *PubSub) notifySubs(src peer.ID, msg *pb.Message) {
 	for _, topic := range msg.GetTopicIDs() {
 		subs := p.myTopics[topic]
 		for f := range subs {
 			select {
-			case f.ch <- &Message{msg}:
+			case f.ch <- &Message{
+				Message:  msg,
+				senderID: src,
+			}:
 			default:
 				log.Infof("Can't deliver message to subscription for topic %s; subscriber too slow", topic)
 			}
@@ -616,7 +624,10 @@ func (p *PubSub) handleIncomingRPC(rpc *RPC) {
 			continue
 		}
 
-		msg := &Message{pmsg}
+		msg := &Message{
+			Message:  pmsg,
+			senderID: rpc.from,
+		}
 		p.pushMsg(rpc.from, msg)
 	}
 
@@ -664,7 +675,7 @@ func (p *PubSub) pushMsg(src peer.ID, msg *Message) {
 }
 
 func (p *PubSub) publishMessage(from peer.ID, pmsg *pb.Message) {
-	p.notifySubs(pmsg)
+	p.notifySubs(from, pmsg)
 	p.rt.Publish(from, pmsg)
 }
 
@@ -747,7 +758,10 @@ func (p *PubSub) Publish(topic string, data []byte) error {
 			return err
 		}
 	}
-	p.publish <- &Message{m}
+	p.publish <- &Message{
+		Message:  m,
+		senderID: p.host.ID(),
+	}
 	return nil
 }