libp2p網路通信中還有一種方式就是PubSub模式，也稱訂閱發布的模式，官方給出了訂閱發布模式的一個案例=> 聊天室

在此學習記錄一下

官方代碼地址：https://github.com/libp2p/go-libp2p/tree/master/examples/pubsub

一、效果演示

二、代碼理解

2.1 總體框架

總的來說代碼構成由這五個步驟：

1~2兩步較為簡單不再贅述, 下面幾點分點描述

2.2 創建mDNS節點發現服務

注意，使用mDNS作為節點發現需要保證所有的節點在同一個局域網即節點發現的范圍在同一個局域網下

// discoveryNotifee gets notified when we find a new peer via mDNS discovery
// 節點發現的通告結構體，繼承Notifee
type discoveryNotifee struct {
	h host.Host
}

// HandlePeerFound connects to peers discovered via mDNS. Once they're connected,
// the PubSub system will automatically start interacting with them if they also
// support PubSub.
// 繼承函式，節點發現后的處理函式：自動鏈接節點
func (n *discoveryNotifee) HandlePeerFound(pi peer.AddrInfo) {
	fmt.Printf("discovered new peer %s\n", pi.ID.Pretty())
	err := n.h.Connect(context.Background(), pi)
	if err != nil {
		fmt.Printf("error connecting to peer %s: %s\n", pi.ID.Pretty(), err)
	}
}

// setupDiscovery creates an mDNS discovery service and attaches it to the libp2p Host.
// This lets us automatically discover peers on the same LAN and connect to them.
func setupDiscovery(ctx context.Context, h host.Host) error {
	// setup mDNS discovery to find local peers
	disc := mdns.NewMdnsService(h, DiscoveryServiceTag)

	n := discoveryNotifee{h: h}
	disc.RegisterNotifee(&n)
	return nil
}

2.3 加入聊天房

每個節點通過訂閱房間的統一topic實作PubSub

1.聊天房資料結構

與一個topic一一對應，可以通過ChatRoom.Publish在topic中發布訊息，并且接收所有的訊息到Messages的channel中，

type ChatRoom struct {
	// Messages is a channel of messages received from other peers in the chat room
	Messages chan *ChatMessage

	ctx   context.Context
	ps    *pubsub.PubSub
	topic *pubsub.Topic
	sub   *pubsub.Subscription

	roomName string
	self     peer.ID
	nick     string
}

// ChatMessage gets converted to/from JSON and sent in the body of pubsub messages.
type ChatMessage struct {
	Message    string
	SenderID   string
	SenderNick string
}

func (cr *ChatRoom) Publish(message string) error {
	m := ChatMessage{
		Message:    message,
		SenderID:   cr.self.Pretty(),
		SenderNick: cr.nick,
	}
	msgBytes, err := json.Marshal(m)
	if err != nil {
		return err
	}
	return cr.topic.Publish(cr.ctx, msgBytes)
}

2.加入聊天房邏輯

分為三步，成功后回傳一個新的ChatRoom實體

func JoinChatRoom(ctx context.Context, ps *pubsub.PubSub, selfID peer.ID, nickname string, roomName string) (*ChatRoom, error) {
	// join the pubsub topic
	topic, err := ps.Join(topicName(roomName))
	if err != nil {
		return nil, err
	}

	// and subscribe to it
	sub, err := topic.Subscribe()
	if err != nil {
		return nil, err
	}

	cr := &ChatRoom{
		ctx:      ctx,
		ps:       ps,
		topic:    topic,
		sub:      sub,
		self:     selfID,
		nick:     nickname,
		roomName: roomName,
		Messages: make(chan *ChatMessage, ChatRoomBufSize),
	}

	// start reading messages from the subscription in a loop
	go cr.readLoop()
	return cr, nil
}

發布和訂閱較為直觀，下面是回圈讀取：

3.回圈讀取訊息內容

回圈讀取內容，并將被容加入到訊息channel中

// readLoop pulls messages from the pubsub topic and pushes them onto the Messages channel.
func (cr *ChatRoom) readLoop() {
	for {
		// Next returns the next message in our subscription
		// 找到下一個訊息
		msg, err := cr.sub.Next(cr.ctx)
		if err != nil {
			close(cr.Messages)
			return
		}
		// only forward messages delivered by others
		// 只接收別人的訊息
		if msg.ReceivedFrom == cr.self {
			continue
		}
		// 反序列化
		cm := new(ChatMessage)
		err = json.Unmarshal(msg.Data, cm)
		if err != nil {
			continue
		}
		// send valid messages onto the Messages channel
		// 把訊息加入 Messages channel
		cr.Messages <- cm
	}
}

4.獲取當前topic所有連接者

func (cr *ChatRoom) ListPeers() []peer.ID {
  // ListPeers returns a list of peers we are connected to in the given topic.
	return cr.ps.ListPeers(topicName(cr.roomName))
}

對于UI部分不是重點，會使用即可

總體來說案例使用較為簡單，可以快速上手！

覺得不錯的話，請點贊關注呦～～你的關注就是博主的動力
關注公眾號，查看更多go開發、密碼學和區塊鏈科研內容：