vendor/github.com/ipfs/go-ipfs-api/pubsub.go

   1 package shell
   2
   3 import (
   4         "encoding/json"
   5
   6         "github.com/libp2p/go-libp2p-peer"
   7 )
   8
   9 // Message is a pubsub message.
  10 type Message struct {
  11         From     peer.ID
  12         Data     []byte
  13         Seqno    []byte
  14         TopicIDs []string
  15 }
  16
  17 // PubSubSubscription allow you to receive pubsub records that where published on the network.
  18 type PubSubSubscription struct {
  19         resp *Response
  20 }
  21
  22 func newPubSubSubscription(resp *Response) *PubSubSubscription {
  23         sub := &PubSubSubscription{
  24                 resp: resp,
  25         }
  26
  27         return sub
  28 }
  29
  30 // Next waits for the next record and returns that.
  31 func (s *PubSubSubscription) Next() (*Message, error) {
  32         if s.resp.Error != nil {
  33                 return nil, s.resp.Error
  34         }
  35
  36         d := json.NewDecoder(s.resp.Output)
  37
  38         var r struct {
  39                 From     []byte   `json:"from,omitempty"`
  40                 Data     []byte   `json:"data,omitempty"`
  41                 Seqno    []byte   `json:"seqno,omitempty"`
  42                 TopicIDs []string `json:"topicIDs,omitempty"`
  43         }
  44
  45         err := d.Decode(&r)
  46         if err != nil {
  47                 return nil, err
  48         }
  49
  50         from, err := peer.IDFromBytes(r.From)
  51         if err != nil {
  52                 return nil, err
  53         }
  54         return &Message{
  55                 From:     from,
  56                 Data:     r.Data,
  57                 Seqno:    r.Seqno,
  58                 TopicIDs: r.TopicIDs,
  59         }, nil
  60 }
  61
  62 // Cancel cancels the given subscription.
  63 func (s *PubSubSubscription) Cancel() error {
  64         if s.resp.Output == nil {
  65                 return nil
  66         }
  67
  68         return s.resp.Output.Close()
  69 }