Some draft stuff for Janos

swarm/network/requests.go

@@ -122,18 +122,6 @@ func addRequester(rs *storage.RequestStatus, req *retrieveRequestMsg) {
 	rs.Requesters[req.Id] = append(list, req)
 }
 
-/*
- store requests are put in netstore so they are stored and then
- forwarded to the peers in their kademlia proximity bin by the syncer
-*/
-type storeRequestMsg struct {
-	Key   storage.Key
-	SData []byte // the stored chunk Data (incl size)
-	// optional
-	Id   uint64 // request ID. if delivery, the ID is retrieve request ID
-	from Peer   // [not serialised] protocol registers the requester
-}
-
 func (self storeRequestMsg) String() string {
 	var from string
 	if self.from == nil {
@@ -147,18 +135,3 @@ func (self storeRequestMsg) String() string {
 	}
 	return fmt.Sprintf("from: %v, ID: %v, SData %x", from, self.Id, self.SData[:end])
 }
-
-// the entrypoint for store requests coming from the bzz wire protocol
-// if key found locally, return. otherwise
-// remote is untrusted, so hash is verified and chunk passed on to NetStore
-func (self *RequestHandler) handleStoreRequestMsg(msg interface{}, p Peer) error {
-	req := msg.(*storeRequestMsg)
-	req.from = p
-	// TODO: chunk validation
-	chunk := storage.NewChunk(req.Key, nil)
-	chunk.SData = req.SData
-	chunk.Source = p
-	self.netStore.Put(chunk)
-	log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, p))
-	return nil
-}

swarm/network/streamer.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
 	bv "github.com/ethereum/go-ethereum/swarm/network/bitvector"
@@ -91,6 +92,18 @@ type UnsyncedKeysMsg struct {
 	*HandoverProof        // HandoverProof
 }
 
+/*
+ store requests are put in netstore so they are stored and then
+ forwarded to the peers in their kademlia proximity bin by the syncer
+*/
+type ChunkDeliveryMsg struct {
+	Key   storage.Key
+	SData []byte // the stored chunk Data (incl size)
+	// optional
+	Id   uint64 // request ID. if delivery, the ID is retrieve request ID
+	from Peer   // [not serialised] protocol registers the requester
+}
+
 // String pretty prints UnsyncedKeysMsg
 func (self UnsyncedKeysMsg) String() string {
 	return fmt.Sprintf("Stream '%v' [%v-%v] (%v)", self.Stream, self.From, self.To, len(self.Hashes)/HashSize)
@@ -170,7 +183,7 @@ func (self *Streamer) PeerInfo(id discover.NodeID) interface{} {
 }
 
 // OutgoingStreamer interface for outgoing peer Streamer
-type OutgoingStreamer interface {
+type OutgoingStreamerBackend interface {
 	CurrentBatch() []byte
 	SetNextBatch(uint64, uint64) ([]byte, uint64, uint64, *HandoverProof, error)
 	GetData([]byte) []byte
@@ -178,7 +191,7 @@ type OutgoingStreamer interface {
 }
 
 // IncomingStreamer interface for incoming peer Streamer
-type IncomingStreamer interface {
+type IncomingStreamerBackend interface {
 	NextBatch(uint64) (uint64, uint64)
 	NeedData([]byte) func()
 	Priority() int
@@ -197,6 +210,16 @@ type StreamerPeer struct {
 	quit         chan struct{}
 }
 
+type IncomingStreamer struct {
+	priority uint8
+	peer     *StreamerPeer
+}
+
+type OutgoingStreamer struct {
+	priority uint8
+	peer     *StreamerPeer
+}
+
 // NewStreamerPeer is the constructor for StreamerPeer
 func NewStreamerPeer(p Peer, streamer *Streamer) *StreamerPeer {
 	self := &StreamerPeer{
@@ -255,6 +278,10 @@ func (self *StreamerPeer) setIncomingStreamer(s Stream, i IncomingStreamer) erro
 	return nil
 }
 
+func (self *OutgoingStreamer) Subscribe(s Stream) OutgoingStreamerBackend {
+
+}
+
 // Subscribe initiates the streamer
 func (self *StreamerPeer) Subscribe(s Stream, from, to uint64) error {
 	f, err := self.streamer.GetIncomingStreamer(s)
@@ -384,6 +411,18 @@ func (self *StreamerPeer) handleTakeoverProofMsg(msg interface{}) error {
 	return nil
 }
 
+func (self *StreamerPeer) handleChunkDeliveryMsg(msg interface{}) error {
+	req := msg.(*chunkDeliveryMsg)
+	req.from = self
+	// TODO: chunk validation
+	chunk := storage.NewChunk(req.Key, nil)
+	chunk.SData = req.SData
+	chunk.Source = p
+	self.netStore.Put(chunk)
+	log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, p))
+	return nil
+}
+
 // Deliver sends a storeRequestMsg protocol message to the peer
 func (self *StreamerPeer) Deliver(chunk *storage.Chunk, priority int) error {
 	msg := &storeRequestMsg{
@@ -451,6 +490,9 @@ func (self *StreamerPeer) HandleMsg(msg interface{}) error {
 	case *WantedKeysMsg:
 		return self.handleWantedKeysMsg(msg)
 
+	case *ChunkDeliveryMsg:
+		return self.handleChunkDeliveryMsg(msg)
+
 	default:
 		return fmt.Errorf("unknown message type: %T", msg)
 	}

swarm/network/syncer.go

@@ -148,6 +148,10 @@ type IncomingSwarmSyncer struct {
 	end, start    uint64
 }
 
+type {
+	Subscribe()
+}
+
 // NewIncomingSwarmSyncer is a contructor for provable data exchange syncer
 func NewIncomingSwarmSyncer(po uint8, priority int, intervals []uint64, p Peer, store storage.ChunkStore, chunker storage.Chunker) (*IncomingSwarmSyncer, error) {
 	self := &IncomingSwarmSyncer{