Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

wire a context in most of the go-ipfs data pipeline, connect them #78

Closed
wants to merge 1 commit into from
Closed

wire a context in most of the go-ipfs data pipeline, connect them #78

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions fetch.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ type fetchProtocol struct {
host host.Host
}

type getValue func(key string) ([]byte, error)
type getValue func(ctx context.Context, key string) ([]byte, error)

func newFetchProtocol(ctx context.Context, host host.Host, getData getValue) *fetchProtocol {
p := &fetchProtocol{ctx, host}
Expand All @@ -46,7 +46,7 @@ func (p *fetchProtocol) receive(s network.Stream, getData getValue) {
return
}

response, err := getData(msg.Identifier)
response, err := getData(p.ctx, msg.Identifier)
var respProto pb.FetchResponse

if err != nil {
Expand Down
2 changes: 1 addition & 1 deletion fetch_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ type datastore struct {
data map[string][]byte
}

func (d *datastore) Lookup(key string) ([]byte, error) {
func (d *datastore) Lookup(ctx context.Context, key string) ([]byte, error) {
v, ok := d.data[key]
if !ok {
return nil, errors.New("key not found")
Expand Down
26 changes: 13 additions & 13 deletions pubsub.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ func (p *PubsubValueStore) PutValue(ctx context.Context, key string, value []byt

ti.dbWriteMx.Lock()
defer ti.dbWriteMx.Unlock()
recCmp, err := p.putLocal(ti, key, value)
recCmp, err := p.putLocal(ctx, ti, key, value)
if err != nil {
return err
}
Expand All @@ -147,12 +147,12 @@ func (p *PubsubValueStore) PutValue(ctx context.Context, key string, value []byt
// First return value is 0 if equal, greater than 0 if better, less than 0 if worse.
// Second return value is true if valid.
//
func (p *PubsubValueStore) compare(key string, val []byte) (int, bool) {
func (p *PubsubValueStore) compare(ctx context.Context, key string, val []byte) (int, bool) {
if p.Validator.Validate(key, val) != nil {
return -1, false
}

old, err := p.getLocal(key)
old, err := p.getLocal(ctx, key)
if err != nil {
// If the old one is invalid, the new one is *always* better.
return 1, true
Expand Down Expand Up @@ -192,7 +192,7 @@ func (p *PubsubValueStore) Subscribe(key string) error {
src peer.ID,
msg *pubsub.Message,
) pubsub.ValidationResult {
cmp, valid := p.compare(key, msg.GetData())
cmp, valid := p.compare(ctx, key, msg.GetData())
if !valid {
return pubsub.ValidationReject
}
Expand Down Expand Up @@ -271,7 +271,7 @@ func (p *PubsubValueStore) rebroadcast(ctx context.Context) {
p.mx.Unlock()
if len(topics) > 0 {
for i, k := range keys {
val, err := p.getLocal(k)
val, err := p.getLocal(ctx, k)
if err == nil {
topic := topics[i].topic
select {
Expand Down Expand Up @@ -300,16 +300,16 @@ func (p *PubsubValueStore) psPublishChannel(ctx context.Context, topic *pubsub.T
// Requires that the ti.dbWriteMx is held when called
// Returns true if the value is better then what is currently in the datastore
// Returns any errors from putting the data in the datastore
func (p *PubsubValueStore) putLocal(ti *topicInfo, key string, value []byte) (int, error) {
cmp, valid := p.compare(key, value)
func (p *PubsubValueStore) putLocal(ctx context.Context, ti *topicInfo, key string, value []byte) (int, error) {
cmp, valid := p.compare(ctx, key, value)
if valid && cmp > 0 {
return cmp, p.ds.Put(dshelp.NewKeyFromBinary([]byte(key)), value)
return cmp, p.ds.Put(ctx, dshelp.NewKeyFromBinary([]byte(key)), value)
}
return cmp, nil
}

func (p *PubsubValueStore) getLocal(key string) ([]byte, error) {
val, err := p.ds.Get(dshelp.NewKeyFromBinary([]byte(key)))
func (p *PubsubValueStore) getLocal(ctx context.Context, key string) ([]byte, error) {
val, err := p.ds.Get(ctx, dshelp.NewKeyFromBinary([]byte(key)))
if err != nil {
// Don't invalidate due to ds errors.
if err == ds.ErrNotFound {
Expand All @@ -330,7 +330,7 @@ func (p *PubsubValueStore) GetValue(ctx context.Context, key string, opts ...rou
return nil, err
}

return p.getLocal(key)
return p.getLocal(ctx, key)
}

func (p *PubsubValueStore) SearchValue(ctx context.Context, key string, opts ...routing.Option) (<-chan []byte, error) {
Expand All @@ -342,7 +342,7 @@ func (p *PubsubValueStore) SearchValue(ctx context.Context, key string, opts ...
defer p.watchLk.Unlock()

out := make(chan []byte, 1)
lv, err := p.getLocal(key)
lv, err := p.getLocal(ctx, key)
if err == nil {
out <- lv
close(out)
Expand Down Expand Up @@ -513,7 +513,7 @@ func (p *PubsubValueStore) handleSubscription(ctx context.Context, ti *topicInfo
}

ti.dbWriteMx.Lock()
recCmp, err := p.putLocal(ti, key, data)
recCmp, err := p.putLocal(ctx, ti, key, data)
ti.dbWriteMx.Unlock()
if recCmp > 0 {
if err != nil {
Expand Down