1 files changed, 0 insertions, 417 deletions
diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go
deleted file mode 100644
index 339101b88..000000000
--- a/swarm/network/stream/messages.go
+++ /dev/null
@@ -1,417 +0,0 @@
-// Copyright 2018 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
-
-package stream
-
-import (
-	"context"
-	"fmt"
-	"time"
-
-	"github.com/ethereum/go-ethereum/metrics"
-	"github.com/ethereum/go-ethereum/swarm/log"
-	bv "github.com/ethereum/go-ethereum/swarm/network/bitvector"
-	"github.com/ethereum/go-ethereum/swarm/storage"
-)
-
-var syncBatchTimeout = 30 * time.Second
-
-// Stream defines a unique stream identifier.
-type Stream struct {
-	// Name is used for Client and Server functions identification.
-	Name string
-	// Key is the name of specific stream data.
-	Key string
-	// Live defines whether the stream delivers only new data
-	// for the specific stream.
-	Live bool
-}
-
-func NewStream(name string, key string, live bool) Stream {
-	return Stream{
-		Name: name,
-		Key:  key,
-		Live: live,
-	}
-}
-
-// String return a stream id based on all Stream fields.
-func (s Stream) String() string {
-	t := "h"
-	if s.Live {
-		t = "l"
-	}
-	return fmt.Sprintf("%s|%s|%s", s.Name, s.Key, t)
-}
-
-// SubcribeMsg is the protocol msg for requesting a stream(section)
-type SubscribeMsg struct {
-	Stream   Stream
-	History  *Range `rlp:"nil"`
-	Priority uint8  // delivered on priority channel
-}
-
-// RequestSubscriptionMsg is the protocol msg for a node to request subscription to a
-// specific stream
-type RequestSubscriptionMsg struct {
-	Stream   Stream
-	History  *Range `rlp:"nil"`
-	Priority uint8  // delivered on priority channel
-}
-
-func (p *Peer) handleRequestSubscription(ctx context.Context, req *RequestSubscriptionMsg) (err error) {
-	log.Debug(fmt.Sprintf("handleRequestSubscription: streamer %s to subscribe to %s with stream %s", p.streamer.addr, p.ID(), req.Stream))
-	if err = p.streamer.Subscribe(p.ID(), req.Stream, req.History, req.Priority); err != nil {
-		// The error will be sent as a subscribe error message
-		// and will not be returned as it will prevent any new message
-		// exchange between peers over p2p. Instead, error will be returned
-		// only if there is one from sending subscribe error message.
-		err = p.Send(ctx, SubscribeErrorMsg{
-			Error: err.Error(),
-		})
-	}
-	return err
-}
-
-func (p *Peer) handleSubscribeMsg(ctx context.Context, req *SubscribeMsg) (err error) {
-	metrics.GetOrRegisterCounter("peer.handlesubscribemsg", nil).Inc(1)
-
-	defer func() {
-		if err != nil {
-			// The error will be sent as a subscribe error message
-			// and will not be returned as it will prevent any new message
-			// exchange between peers over p2p. Instead, error will be returned
-			// only if there is one from sending subscribe error message.
-			err = p.Send(context.TODO(), SubscribeErrorMsg{
-				Error: err.Error(),
-			})
-		}
-	}()
-
-	log.Debug("received subscription", "from", p.streamer.addr, "peer", p.ID(), "stream", req.Stream, "history", req.History)
-
-	f, err := p.streamer.GetServerFunc(req.Stream.Name)
-	if err != nil {
-		return err
-	}
-
-	s, err := f(p, req.Stream.Key, req.Stream.Live)
-	if err != nil {
-		return err
-	}
-	os, err := p.setServer(req.Stream, s, req.Priority)
-	if err != nil {
-		return err
-	}
-
-	var from uint64
-	var to uint64
-	if !req.Stream.Live && req.History != nil {
-		from = req.History.From
-		to = req.History.To
-	}
-
-	go func() {
-		if err := p.SendOfferedHashes(os, from, to); err != nil {
-			log.Warn("SendOfferedHashes error", "peer", p.ID().TerminalString(), "err", err)
-		}
-	}()
-
-	if req.Stream.Live && req.History != nil {
-		// subscribe to the history stream
-		s, err := f(p, req.Stream.Key, false)
-		if err != nil {
-			return err
-		}
-
-		os, err := p.setServer(getHistoryStream(req.Stream), s, getHistoryPriority(req.Priority))
-		if err != nil {
-			return err
-		}
-		go func() {
-			if err := p.SendOfferedHashes(os, req.History.From, req.History.To); err != nil {
-				log.Warn("SendOfferedHashes error", "peer", p.ID().TerminalString(), "err", err)
-			}
-		}()
-	}
-
-	return nil
-}
-
-type SubscribeErrorMsg struct {
-	Error string
-}
-
-func (p *Peer) handleSubscribeErrorMsg(req *SubscribeErrorMsg) (err error) {
-	//TODO the error should be channeled to whoever calls the subscribe
-	return fmt.Errorf("subscribe to peer %s: %v", p.ID(), req.Error)
-}
-
-type UnsubscribeMsg struct {
-	Stream Stream
-}
-
-func (p *Peer) handleUnsubscribeMsg(req *UnsubscribeMsg) error {
-	return p.removeServer(req.Stream)
-}
-
-type QuitMsg struct {
-	Stream Stream
-}
-
-func (p *Peer) handleQuitMsg(req *QuitMsg) error {
-	err := p.removeClient(req.Stream)
-	if _, ok := err.(*notFoundError); ok {
-		return nil
-	}
-	return err
-}
-
-// OfferedHashesMsg is the protocol msg for offering to hand over a
-// stream section
-type OfferedHashesMsg struct {
-	Stream         Stream // name of Stream
-	From, To       uint64 // peer and db-specific entry count
-	Hashes         []byte // stream of hashes (128)
-	*HandoverProof        // HandoverProof
-}
-
-// String pretty prints OfferedHashesMsg
-func (m OfferedHashesMsg) String() string {
-	return fmt.Sprintf("Stream '%v' [%v-%v] (%v)", m.Stream, m.From, m.To, len(m.Hashes)/HashSize)
-}
-
-// handleOfferedHashesMsg protocol msg handler calls the incoming streamer interface
-// Filter method
-func (p *Peer) handleOfferedHashesMsg(ctx context.Context, req *OfferedHashesMsg) error {
-	metrics.GetOrRegisterCounter("peer.handleofferedhashes", nil).Inc(1)
-
-	c, _, err := p.getOrSetClient(req.Stream, req.From, req.To)
-	if err != nil {
-		return err
-	}
-
-	hashes := req.Hashes
-	lenHashes := len(hashes)
-	if lenHashes%HashSize != 0 {
-		return fmt.Errorf("error invalid hashes length (len: %v)", lenHashes)
-	}
-
-	want, err := bv.New(lenHashes / HashSize)
-	if err != nil {
-		return fmt.Errorf("error initiaising bitvector of length %v: %v", lenHashes/HashSize, err)
-	}
-
-	var wantDelaySet bool
-	var wantDelay time.Time
-
-	ctr := 0
-	errC := make(chan error)
-	ctx, cancel := context.WithTimeout(ctx, syncBatchTimeout)
-
-	ctx = context.WithValue(ctx, "source", p.ID().String())
-	for i := 0; i < lenHashes; i += HashSize {
-		hash := hashes[i : i+HashSize]
-
-		if wait := c.NeedData(ctx, hash); wait != nil {
-			ctr++
-			want.Set(i/HashSize, true)
-
-			// measure how long it takes before we mark chunks for retrieval, and actually send the request
-			if !wantDelaySet {
-				wantDelaySet = true
-				wantDelay = time.Now()
-			}
-
-			// create request and wait until the chunk data arrives and is stored
-			go func(w func(context.Context) error) {
-				select {
-				case errC <- w(ctx):
-				case <-ctx.Done():
-				}
-			}(wait)
-		}
-	}
-
-	go func() {
-		defer cancel()
-		for i := 0; i < ctr; i++ {
-			select {
-			case err := <-errC:
-				if err != nil {
-					log.Debug("client.handleOfferedHashesMsg() error waiting for chunk, dropping peer", "peer", p.ID(), "err", err)
-					p.Drop()
-					return
-				}
-			case <-ctx.Done():
-				log.Debug("client.handleOfferedHashesMsg() context done", "ctx.Err()", ctx.Err())
-				return
-			case <-c.quit:
-				log.Debug("client.handleOfferedHashesMsg() quit")
-				return
-			}
-		}
-		select {
-		case c.next <- c.batchDone(p, req, hashes):
-		case <-c.quit:
-			log.Debug("client.handleOfferedHashesMsg() quit")
-		case <-ctx.Done():
-			log.Debug("client.handleOfferedHashesMsg() context done", "ctx.Err()", ctx.Err())
-		}
-	}()
-	// only send wantedKeysMsg if all missing chunks of the previous batch arrived
-	// except
-	if c.stream.Live {
-		c.sessionAt = req.From
-	}
-	from, to := c.nextBatch(req.To + 1)
-	log.Trace("set next batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To, "addr", p.streamer.addr)
-	if from == to {
-		return nil
-	}
-
-	msg := &WantedHashesMsg{
-		Stream: req.Stream,
-		Want:   want.Bytes(),
-		From:   from,
-		To:     to,
-	}
-
-	log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To)
-	select {
-	case err := <-c.next:
-		if err != nil {
-			log.Warn("c.next error dropping peer", "err", err)
-			p.Drop()
-			return err
-		}
-	case <-c.quit:
-		log.Debug("client.handleOfferedHashesMsg() quit")
-		return nil
-	case <-ctx.Done():
-		log.Debug("client.handleOfferedHashesMsg() context done", "ctx.Err()", ctx.Err())
-		return nil
-	}
-	log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To)
-
-	// record want delay
-	if wantDelaySet {
-		metrics.GetOrRegisterResettingTimer("handleoffered.wantdelay", nil).UpdateSince(wantDelay)
-	}
-
-	err = p.SendPriority(ctx, msg, c.priority)
-	if err != nil {
-		log.Warn("SendPriority error", "err", err)
-	}
-
-	return nil
-}
-
-// WantedHashesMsg is the protocol msg data for signaling which hashes
-// offered in OfferedHashesMsg downstream peer actually wants sent over
-type WantedHashesMsg struct {
-	Stream   Stream
-	Want     []byte // bitvector indicating which keys of the batch needed
-	From, To uint64 // next interval offset - empty if not to be continued
-}
-
-// String pretty prints WantedHashesMsg
-func (m WantedHashesMsg) String() string {
-	return fmt.Sprintf("Stream '%v', Want: %x, Next: [%v-%v]", m.Stream, m.Want, m.From, m.To)
-}
-
-// handleWantedHashesMsg protocol msg handler
-// * sends the next batch of unsynced keys
-// * sends the actual data chunks as per WantedHashesMsg
-func (p *Peer) handleWantedHashesMsg(ctx context.Context, req *WantedHashesMsg) error {
-	metrics.GetOrRegisterCounter("peer.handlewantedhashesmsg", nil).Inc(1)
-
-	log.Trace("received wanted batch", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To)
-	s, err := p.getServer(req.Stream)
-	if err != nil {
-		return err
-	}
-	hashes := s.currentBatch
-	// launch in go routine since GetBatch blocks until new hashes arrive
-	go func() {
-		if err := p.SendOfferedHashes(s, req.From, req.To); err != nil {
-			log.Warn("SendOfferedHashes error", "peer", p.ID().TerminalString(), "err", err)
-		}
-	}()
-	// go p.SendOfferedHashes(s, req.From, req.To)
-	l := len(hashes) / HashSize
-
-	log.Trace("wanted batch length", "peer", p.ID(), "stream", req.Stream, "from", req.From, "to", req.To, "lenhashes", len(hashes), "l", l)
-	want, err := bv.NewFromBytes(req.Want, l)
-	if err != nil {
-		return fmt.Errorf("error initiaising bitvector of length %v: %v", l, err)
-	}
-	for i := 0; i < l; i++ {
-		if want.Get(i) {
-			metrics.GetOrRegisterCounter("peer.handlewantedhashesmsg.actualget", nil).Inc(1)
-
-			hash := hashes[i*HashSize : (i+1)*HashSize]
-			data, err := s.GetData(ctx, hash)
-			if err != nil {
-				return fmt.Errorf("handleWantedHashesMsg get data %x: %v", hash, err)
-			}
-			chunk := storage.NewChunk(hash, data)
-			syncing := true
-			if err := p.Deliver(ctx, chunk, s.priority, syncing); err != nil {
-				return err
-			}
-		}
-	}
-	return nil
-}
-
-// Handover represents a statement that the upstream peer hands over the stream section
-type Handover struct {
-	Stream     Stream // name of stream
-	Start, End uint64 // index of hashes
-	Root       []byte // Root hash for indexed segment inclusion proofs
-}
-
-// HandoverProof represents a signed statement that the upstream peer handed over the stream section
-type HandoverProof struct {
-	Sig []byte // Sign(Hash(Serialisation(Handover)))
-	*Handover
-}
-
-// Takeover represents a statement that downstream peer took over (stored all data)
-// handed over
-type Takeover Handover
-
-//  TakeoverProof represents a signed statement that the downstream peer took over
-// the stream section
-type TakeoverProof struct {
-	Sig []byte // Sign(Hash(Serialisation(Takeover)))
-	*Takeover
-}
-
-// TakeoverProofMsg is the protocol msg sent by downstream peer
-type TakeoverProofMsg TakeoverProof
-
-// String pretty prints TakeoverProofMsg
-func (m TakeoverProofMsg) String() string {
-	return fmt.Sprintf("Stream: '%v' [%v-%v], Root: %x, Sig: %x", m.Stream, m.Start, m.End, m.Root, m.Sig)
-}
-
-func (p *Peer) handleTakeoverProofMsg(ctx context.Context, req *TakeoverProofMsg) error {
-	_, err := p.getServer(req.Stream)
-	// store the strongest takeoverproof for the stream in streamer
-	return err
-}