pubsub.go

package p2pd

import (
	pb "github.com/learning-at-home/go-libp2p-daemon/pb"

	"github.com/libp2p/go-libp2p/core/peer"

	ggio "github.com/gogo/protobuf/io"
	ps "github.com/libp2p/go-libp2p-pubsub"
)

func (d *Daemon) doPubsub(req *pb.Request) (*pb.Response, *ps.Subscription) {
	if d.pubsub == nil {
		return errorResponseString("PubSub not enabled"), nil
	}

	if req.Pubsub == nil {
		return errorResponseString("Malformed request; missing parameters"), nil
	}

	switch req.Pubsub.GetType() {
	case pb.PSRequest_GET_TOPICS:
		return d.doPubsubGetTopics(req.Pubsub)

	case pb.PSRequest_LIST_PEERS:
		return d.doPubsubListPeers(req.Pubsub)

	case pb.PSRequest_PUBLISH:
		return d.doPubsubPublish(req.Pubsub)

	case pb.PSRequest_SUBSCRIBE:
		return d.doPubsubSubscribe(req.Pubsub)

	default:
		log.Debugw("unexpected pubsub request type", "type", req.Pubsub.GetType())
		return errorResponseString("Unexpected request"), nil
	}
}

func (d *Daemon) doPubsubGetTopics(req *pb.PSRequest) (*pb.Response, *ps.Subscription) {
	topics := d.pubsub.GetTopics()
	return psOkResponse(psResponseTopics(topics)), nil
}

func (d *Daemon) doPubsubListPeers(req *pb.PSRequest) (*pb.Response, *ps.Subscription) {
	if req.Topic == nil {
		return errorResponseString("Malformed request; missing topic parameter"), nil
	}

	peers := d.pubsub.ListPeers(*req.Topic)
	return psOkResponse(psResponsePeers(peers)), nil
}

func (d *Daemon) doPubsubPublish(req *pb.PSRequest) (*pb.Response, *ps.Subscription) {
	if req.Topic == nil {
		return errorResponseString("Malformed request; missing topic parameter"), nil
	}

	//lint:ignore SA1019 requires API changes
	err := d.pubsub.Publish(*req.Topic, req.Data)
	if err != nil {
		return errorResponse(err), nil
	}

	return okResponse(), nil
}

func (d *Daemon) doPubsubSubscribe(req *pb.PSRequest) (*pb.Response, *ps.Subscription) {
	if req.Topic == nil {
		return errorResponseString("Malformed request; missing topic parameter"), nil
	}

	//lint:ignore SA1019 requires API changes
	sub, err := d.pubsub.Subscribe(*req.Topic)
	if err != nil {
		return errorResponse(err), nil
	}

	return okResponse(), sub
}

func (d *Daemon) doPubsubPipe(sub *ps.Subscription, r ggio.ReadCloser, w ggio.WriteCloser) {
	go func() {
		// read something until the client closes the connection
		// at which point we cancel the subscription
		for {
			var req pb.Request
			err := r.ReadMsg(&req)
			if err != nil {
				sub.Cancel()
				return
			}

			log.Warnw("unexpected message", "type", req.GetType())
		}
	}()

	for {
		msg, err := sub.Next(d.ctx)
		if err != nil {
			log.Warnw("subscription error", "error", err)
			// goroutine will cancel the subscription once the connection is closed on return
			return
		}

		psmsg := psMessage(msg)
		err = w.WriteMsg(psmsg)
		if err != nil {
			log.Warnw("error writing pubsub message", "error", err)
			// goroutine will cancel the subscription once the connection is closed on return
			return
		}
	}
}

func psResponseTopics(topics []string) *pb.PSResponse {
	return &pb.PSResponse{Topics: topics}
}

func psResponsePeers(peers []peer.ID) *pb.PSResponse {
	xpeers := make([][]byte, len(peers))
	for x, p := range peers {
		xpeers[x] = []byte(p)
	}
	return &pb.PSResponse{PeerIDs: xpeers}
}

func psMessage(msg *ps.Message) *pb.PSMessage {
	return &pb.PSMessage{
		From:      msg.From,
		Data:      msg.Data,
		Seqno:     msg.Seqno,
		TopicIDs:  []string{*msg.Topic},
		Signature: msg.Signature,
		Key:       msg.Key,
	}
}

func psOkResponse(r *pb.PSResponse) *pb.Response {
	res := okResponse()
	res.Pubsub = r
	return res
}