This repository has been archived by the owner on Aug 2, 2021. It is now read-only.
network/newstream: new stream! protocol base implementation #1500
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| // Copyright 2019 The Swarm Authors | ||
| // This file is part of the Swarm library. | ||
| // | ||
| // The Swarm 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 Swarm 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 Swarm library. If not, see <http://www.gnu.org/licenses/>. | ||
|
|
||
| package newstream | ||
|
|
||
| import ( | ||
| "flag" | ||
| "io/ioutil" | ||
| "os" | ||
|
|
||
| "github.com/ethereum/go-ethereum/common" | ||
| "github.com/ethereum/go-ethereum/log" | ||
| "github.com/ethereum/go-ethereum/p2p/enode" | ||
| "github.com/ethersphere/swarm/network" | ||
| "github.com/ethersphere/swarm/storage/localstore" | ||
| "github.com/ethersphere/swarm/storage/mock" | ||
| ) | ||
|
|
||
| var ( | ||
| loglevel = flag.Int("loglevel", 5, "verbosity of logs") | ||
| ) | ||
|
|
||
| func init() { | ||
| flag.Parse() | ||
|
|
||
| log.PrintOrigins(true) | ||
| log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(false)))) | ||
| } | ||
|
|
||
| func newTestLocalStore(id enode.ID, addr *network.BzzAddr, globalStore mock.GlobalStorer) (localStore *localstore.DB, cleanup func(), err error) { | ||
| dir, err := ioutil.TempDir("", "swarm-stream-") | ||
| if err != nil { | ||
| return nil, nil, err | ||
| } | ||
| cleanup = func() { | ||
| os.RemoveAll(dir) | ||
| } | ||
|
|
||
| var mockStore *mock.NodeStore | ||
| if globalStore != nil { | ||
| mockStore = globalStore.NewNodeStore(common.BytesToAddress(id.Bytes())) | ||
| } | ||
|
|
||
| localStore, err = localstore.New(dir, addr.Over(), &localstore.Options{ | ||
| MockStore: mockStore, | ||
| }) | ||
| if err != nil { | ||
| cleanup() | ||
| return nil, nil, err | ||
| } | ||
| return localStore, cleanup, nil | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| // Copyright 2019 The Swarm Authors | ||
| // This file is part of the Swarm library. | ||
| // | ||
| // The Swarm 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 Swarm 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 Swarm library. If not, see <http://www.gnu.org/licenses/>. | ||
|
|
||
| package newstream | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/ethersphere/swarm/chunk" | ||
| "github.com/ethersphere/swarm/network" | ||
| "github.com/ethersphere/swarm/network/bitvector" | ||
| "github.com/ethersphere/swarm/state" | ||
| ) | ||
|
|
||
| var ErrEmptyBatch = errors.New("empty batch") | ||
|
|
||
| const ( | ||
| HashSize = 32 | ||
| BatchSize = 16 | ||
| ) | ||
|
|
||
| // Peer is the Peer extension for the streaming protocol | ||
| type Peer struct { | ||
| *network.BzzPeer | ||
| mtx sync.Mutex | ||
| providers map[string]StreamProvider | ||
| intervalsStore state.Store | ||
|
|
||
| streamCursorsMu sync.Mutex | ||
| streamCursors map[string]uint64 // key: Stream ID string representation, value: session cursor. Keeps cursors for all streams. when unset - we are not interested in that bin | ||
| dirtyStreams map[string]bool // key: stream ID, value: whether cursors for a stream should be updated | ||
| activeBoundedGets map[string]chan struct{} | ||
| openWants map[uint]*want // maintain open wants on the client side | ||
| openOffers map[uint]offer // maintain open offers on the server side | ||
| quit chan struct{} // closed when peer is going offline | ||
| } | ||
|
|
||
| // NewPeer is the constructor for Peer | ||
| func NewPeer(peer *network.BzzPeer, i state.Store, providers map[string]StreamProvider) *Peer { | ||
| p := &Peer{ | ||
| BzzPeer: peer, | ||
| providers: providers, | ||
| intervalsStore: i, | ||
| streamCursors: make(map[string]uint64), | ||
| dirtyStreams: make(map[string]bool), | ||
| openWants: make(map[uint]*want), | ||
| openOffers: make(map[uint]offer), | ||
| quit: make(chan struct{}), | ||
| } | ||
| return p | ||
| } | ||
| func (p *Peer) Left() { | ||
|
There was a problem hiding this comment. This is used at one place. Why not directly inline |
||
| close(p.quit) | ||
| } | ||
|
|
||
| // HandleMsg is the message handler that delegates incoming messages | ||
| func (p *Peer) HandleMsg(ctx context.Context, msg interface{}) error { | ||
| switch msg := msg.(type) { | ||
| default: | ||
| return fmt.Errorf("unknown message type: %T", msg) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| type offer struct { | ||
| ruid uint | ||
| stream ID | ||
| hashes []byte | ||
| requested time.Time | ||
| } | ||
|
|
||
| type want struct { | ||
| ruid uint | ||
| from uint64 | ||
| to uint64 | ||
| stream ID | ||
| hashes map[string]bool | ||
| bv *bitvector.BitVector | ||
| requested time.Time | ||
| remaining uint64 | ||
| chunks chan chunk.Chunk | ||
| done chan error | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why do we need a separate
common_testfile?