using protobuf now, with a vengeance

.gx/lastpubver

-0.1.0: QmUvA4goPKAHLrTcePzzCDHqCpqtzdKAVdMPdrdw45Z2FM
+0.2.0: QmTP5RSsmZacEFanuBhnDE3RC3aqGAoozF9ZmKqcHa8EKH

floodsub.go

 package floodsub
 
 import (
+	"bufio"
 	"fmt"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	pb "github.com/whyrusleeping/go-floodsub/pb"
 
 	ggio "github.com/gogo/protobuf/io"
 	proto "github.com/gogo/protobuf/proto"
-	logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
-	peer "gx/ipfs/QmWtbQU15LaB5B1JC2F7TV9P4K88vD3PpA4AJrwfCjhML8/go-libp2p-peer"
-	host "gx/ipfs/Qmf4ETeAWXuThBfWwonVyFqGFSgTWepUDEr1txcctvpTXS/go-libp2p/p2p/host"
-	inet "gx/ipfs/Qmf4ETeAWXuThBfWwonVyFqGFSgTWepUDEr1txcctvpTXS/go-libp2p/p2p/net"
-	protocol "gx/ipfs/Qmf4ETeAWXuThBfWwonVyFqGFSgTWepUDEr1txcctvpTXS/go-libp2p/p2p/protocol"
+	peer "github.com/ipfs/go-libp2p-peer"
+	logging "github.com/ipfs/go-log"
+	host "github.com/libp2p/go-libp2p/p2p/host"
+	inet "github.com/libp2p/go-libp2p/p2p/net"
+	protocol "github.com/libp2p/go-libp2p/p2p/protocol"
 )
 
-var messageCount uint64
-
 const ID = protocol.ID("/floodsub/1.0.0")
 
 var (
@@ -113,15 +111,14 @@ func (p *PubSub) handleNewStream(s inet.Stream) {
 
 func (p *PubSub) handleSendingMessages(s inet.Stream, in <-chan *RPC) {
 	var dead bool
-	wc := ggio.NewDelimitedWriter(s)
+	bufw := bufio.NewWriter(s)
+	wc := ggio.NewDelimitedWriter(bufw)
 	defer wc.Close()
 	for rpc := range in {
 		if dead {
 			continue
 		}
 
-		atomic.AddUint64(&messageCount, 1)
-
 		err := wc.WriteMsg(&rpc.RPC)
 		if err != nil {
 			log.Errorf("writing message to %s: %s", s.Conn().RemotePeer(), err)
@@ -130,6 +127,15 @@ func (p *PubSub) handleSendingMessages(s inet.Stream, in <-chan *RPC) {
 				p.peerDead <- s.Conn().RemotePeer()
 			}()
 		}
+
+		err = bufw.Flush()
+		if err != nil {
+			log.Errorf("writing message to %s: %s", s.Conn().RemotePeer(), err)
+			dead = true
+			go func() {
+				p.peerDead <- s.Conn().RemotePeer()
+			}()
+		}
 	}
 }
 

floodsub_test.go

@@ -8,8 +8,8 @@ import (
 	"testing"
 	"time"
 
-	host "gx/ipfs/Qmf4ETeAWXuThBfWwonVyFqGFSgTWepUDEr1txcctvpTXS/go-libp2p/p2p/host"
-	netutil "gx/ipfs/Qmf4ETeAWXuThBfWwonVyFqGFSgTWepUDEr1txcctvpTXS/go-libp2p/p2p/test/util"
+	host "github.com/libp2p/go-libp2p/p2p/host"
+	netutil "github.com/libp2p/go-libp2p/p2p/test/util"
 )
 
 func getNetHosts(t *testing.T, n int) []host.Host {
@@ -94,6 +94,4 @@ func TestBasicFloodsub(t *testing.T) {
 			}
 		}
 	}
-
-	fmt.Println("Total Sent Messages: ", messageCount)
 }

notify.go

@@ -3,8 +3,8 @@ package floodsub
 import (
 	"context"
 
-	ma "gx/ipfs/QmYzDkkgAEmrcNzFCiYo6L1dTX4EAG1gZkbtdbd9trL4vd/go-multiaddr"
-	inet "gx/ipfs/Qmf4ETeAWXuThBfWwonVyFqGFSgTWepUDEr1txcctvpTXS/go-libp2p/p2p/net"
+	ma "github.com/jbenet/go-multiaddr"
+	inet "github.com/libp2p/go-libp2p/p2p/net"
 )
 
 var _ inet.Notifiee = (*PubSub)(nil)

package.json

@@ -12,11 +12,17 @@
       "hash": "Qmf4ETeAWXuThBfWwonVyFqGFSgTWepUDEr1txcctvpTXS",
       "name": "go-libp2p",
       "version": "3.4.1"
+    },
+    {
+      "author": "whyrusleeping",
+      "hash": "QmZ4Qi3GaRbjcx28Sme5eMH7RQjGkt8wHxt2a65oLaeFEV",
+      "name": "gogo-protobuf",
+      "version": "0.0.0"
     }
   ],
   "gxVersion": "0.9.0",
   "language": "go",
   "license": "",
   "name": "floodsub",
-  "version": "0.1.0"
+  "version": "0.2.0"
 }

pb/rpc.pb.go

@@ -23,12 +23,6 @@ var _ = proto.Marshal
 var _ = fmt.Errorf
 var _ = math.Inf
 
-// This is a compile-time assertion to ensure that this generated file
-// is compatible with the proto package it is being compiled against.
-// A compilation error at this line likely means your copy of the
-// proto package needs to be updated.
-const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package
-
 type RPC struct {
 	Type             *string  `protobuf:"bytes,1,opt,name=type" json:"type,omitempty"`
 	Topics           []string `protobuf:"bytes,2,rep,name=topics" json:"topics,omitempty"`
@@ -36,10 +30,9 @@ type RPC struct {
 	XXX_unrecognized []byte   `json:"-"`
 }
 
-func (m *RPC) Reset()                    { *m = RPC{} }
-func (m *RPC) String() string            { return proto.CompactTextString(m) }
-func (*RPC) ProtoMessage()               {}
-func (*RPC) Descriptor() ([]byte, []int) { return fileDescriptorRpc, []int{0} }
+func (m *RPC) Reset()         { *m = RPC{} }
+func (m *RPC) String() string { return proto.CompactTextString(m) }
+func (*RPC) ProtoMessage()    {}
 
 func (m *RPC) GetType() string {
 	if m != nil && m.Type != nil {
@@ -70,10 +63,9 @@ type Message struct {
 	XXX_unrecognized []byte  `json:"-"`
 }
 
-func (m *Message) Reset()                    { *m = Message{} }
-func (m *Message) String() string            { return proto.CompactTextString(m) }
-func (*Message) ProtoMessage()               {}
-func (*Message) Descriptor() ([]byte, []int) { return fileDescriptorRpc, []int{1} }
+func (m *Message) Reset()         { *m = Message{} }
+func (m *Message) String() string { return proto.CompactTextString(m) }
+func (*Message) ProtoMessage()    {}
 
 func (m *Message) GetFrom() string {
 	if m != nil && m.From != nil {
@@ -107,19 +99,3 @@ func init() {
 	proto.RegisterType((*RPC)(nil), "floodsub.pb.RPC")
 	proto.RegisterType((*Message)(nil), "floodsub.pb.Message")
 }
-
-func init() { proto.RegisterFile("rpc.proto", fileDescriptorRpc) }
-
-var fileDescriptorRpc = []byte{
-	// 158 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xe2, 0xe2, 0x2c, 0x2a, 0x48, 0xd6,
-	0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0xe2, 0x4e, 0xcb, 0xc9, 0xcf, 0x4f, 0x29, 0x2e, 0x4d, 0xd2,
-	0x2b, 0x48, 0x52, 0x72, 0xe3, 0x62, 0x0e, 0x0a, 0x70, 0x16, 0xe2, 0xe1, 0x62, 0x29, 0xa9, 0x2c,
-	0x48, 0x95, 0x60, 0x54, 0x60, 0xd4, 0xe0, 0x14, 0xe2, 0xe3, 0x62, 0x2b, 0xc9, 0x2f, 0xc8, 0x4c,
-	0x2e, 0x96, 0x60, 0x52, 0x60, 0xd6, 0xe0, 0x14, 0x52, 0xe4, 0x62, 0xce, 0x2d, 0x4e, 0x97, 0x60,
-	0x56, 0x60, 0xd4, 0xe0, 0x36, 0x12, 0xd1, 0x43, 0xd2, 0xaf, 0xe7, 0x9b, 0x5a, 0x5c, 0x9c, 0x98,
-	0x9e, 0xaa, 0xe4, 0xcc, 0xc5, 0x0e, 0x65, 0x82, 0xcc, 0x4a, 0x2b, 0xca, 0xcf, 0x85, 0x9a, 0xc5,
-	0xc3, 0xc5, 0x92, 0x92, 0x58, 0x92, 0x28, 0xc1, 0xa4, 0xc0, 0xa8, 0xc1, 0x23, 0xc4, 0xcb, 0xc5,
-	0x5a, 0x9c, 0x5a, 0x98, 0x97, 0x0f, 0x36, 0x8b, 0x05, 0xc4, 0x05, 0x5b, 0x24, 0xc1, 0x02, 0x52,
-	0x0b, 0x08, 0x00, 0x00, 0xff, 0xff, 0x08, 0xfe, 0xc0, 0xc1, 0xa5, 0x00, 0x00, 0x00,
-}