-
Notifications
You must be signed in to change notification settings - Fork 66
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(p2p update): initial p2p update
this comes with a few goodies. I've moved the logic from Add that leverages the cafs.Store.Fetcher interface into a base package function called FetchDataset. This is now used in both add and remote Updates. I've also bumped tests in a few spots, with more tests to write.
- Loading branch information
Showing
6 changed files
with
242 additions
and
33 deletions.
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
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,92 @@ | ||
| package p2p | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "fmt" | ||
|
|
||
| "github.com/qri-io/qri/base" | ||
| "github.com/qri-io/qri/repo" | ||
| ) | ||
|
|
||
| // MtLogDiff gets info on a dataset | ||
| const MtLogDiff = MsgType("log_diff") | ||
|
|
||
| // RequestLogDiff fetches info about a dataset from qri peers | ||
| // It's expected the local peer has attempted to canonicalize the reference | ||
| // before sending to the network | ||
| func (n *QriNode) RequestLogDiff(ref *repo.DatasetRef) (ldr base.LogDiffResult, err error) { | ||
| log.Debugf("%s RequestLogDiff %s", n.ID, ref) | ||
|
|
||
| p, err := n.ConnectToPeer(n.ctx, PeerConnectionParams{ | ||
| Peername: ref.Peername, | ||
| ProfileID: ref.ProfileID, | ||
| }) | ||
|
|
||
| if err != nil { | ||
| err = fmt.Errorf("coudn't connection to peer: %s", err.Error()) | ||
| return | ||
| } | ||
|
|
||
| // TODO - deal with max limit / offset / pagination issuez | ||
| rLog, err := base.DatasetLog(n.Repo, *ref, 10000, 0, false) | ||
| if err != nil { | ||
| return | ||
| } | ||
|
|
||
| replies := make(chan Message) | ||
| req, err := NewJSONBodyMessage(n.ID, MtLogDiff, rLog) | ||
| req = req.WithHeaders("phase", "request") | ||
| if err != nil { | ||
| log.Debug(err.Error()) | ||
| return | ||
| } | ||
|
|
||
| for _, pid := range p.PeerIDs { | ||
| if err = n.SendMessage(req, replies, pid); err != nil { | ||
| log.Debugf("%s err: %s", pid, err.Error()) | ||
| continue | ||
| } | ||
|
|
||
| res := <-replies | ||
| ldr = base.LogDiffResult{} | ||
| if err = json.Unmarshal(res.Body, &ldr); err == nil { | ||
| break | ||
| } | ||
| } | ||
|
|
||
| return | ||
| } | ||
|
|
||
| func (n *QriNode) handleLogDiff(ws *WrappedStream, msg Message) (hangup bool) { | ||
| hangup = true | ||
|
|
||
| switch msg.Header("phase") { | ||
| case "request": | ||
| remoteLog := []repo.DatasetRef{} | ||
| if err := json.Unmarshal(msg.Body, &remoteLog); err != nil { | ||
| log.Debug(err.Error()) | ||
| return | ||
| } | ||
|
|
||
| res := msg | ||
| ldr, err := base.LogDiff(n.Repo, remoteLog) | ||
| if err != nil { | ||
| log.Error(err) | ||
| return | ||
| } | ||
|
|
||
| res, err = msg.UpdateJSON(ldr) | ||
| if err != nil { | ||
| log.Error(err) | ||
| return | ||
| } | ||
|
|
||
| res = res.WithHeaders("phase", "response") | ||
| if err := ws.sendMessage(res); err != nil { | ||
| log.Debug(err.Error()) | ||
| return | ||
| } | ||
| } | ||
|
|
||
| return | ||
| } |
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,73 @@ | ||
| package p2p | ||
|
|
||
| import ( | ||
| "testing" | ||
| ) | ||
|
|
||
| func TestRequestLogDiff(t *testing.T) { | ||
| // ctx := context.Background() | ||
| // streams := ioes.NewDiscardIOStreams() | ||
| // factory := p2ptest.NewTestNodeFactory(NewTestableQriNode) | ||
| // testPeers, err := p2ptest.NewTestDirNetwork(ctx, factory) | ||
| // if err != nil { | ||
| // t.Fatalf("error creating network: %s", err.Error()) | ||
| // } | ||
| // if err := p2ptest.ConnectQriNodes(ctx, testPeers); err != nil { | ||
| // t.Fatalf("error connecting peers: %s", err.Error()) | ||
| // } | ||
|
|
||
| // peers := asQriNodes(testPeers) | ||
|
|
||
| // tc, err := dstest.NewTestCaseFromDir("testdata/tim/craigslist") | ||
| // if err != nil { | ||
| // t.Fatal(err) | ||
| // } | ||
|
|
||
| // // add a dataset to tim | ||
| // ref, _, err := base.CreateDataset(peers[4].Repo, streams, tc.Name, tc.Input, tc.BodyFile(), false, true) | ||
| // if err != nil { | ||
| // t.Fatal(err) | ||
| // } | ||
|
|
||
| // // | ||
| // // peers[] | ||
|
|
||
| // prevTitle := tc.Input.Meta.Title | ||
| // tc.Input.Meta.Title = "update" | ||
| // tc.Input.PreviousPath = ref.Path | ||
| // defer func() { | ||
| // // because test cases are cached for performance, we need to clean up any mutation to | ||
| // // testcase input | ||
| // tc.Input.Meta.Title = prevTitle | ||
| // tc.Input.PreviousPath = ref.Path | ||
| // }() | ||
|
|
||
| // ref2, _, err := base.CreateDataset(peers[4].Repo, streams, tc.Name, tc.Input, tc.BodyFile(), false, true) | ||
| // if err != nil { | ||
| // t.Fatal(err) | ||
| // } | ||
|
|
||
| // t.Logf("testing RequestDatasetLog message with %d peers", len(peers)) | ||
| // // var wg sync.WaitGroup | ||
| // for i, p1 := range peers { | ||
| // for _, p2 := range peers[i+1:] { | ||
| // // TODO - having these in parallel is causing races when encoding logs | ||
| // // wg.Add(1) | ||
| // // go func(p1, p2 *QriNode) { | ||
| // // defer wg.Done() | ||
|
|
||
| // refs, err := p1.RequestDatasetLog(ref, 100, 0) | ||
| // if err != nil { | ||
| // t.Errorf("%s -> %s error: %s", p1.ID.Pretty(), p2.ID.Pretty(), err.Error()) | ||
| // } | ||
| // if refs == nil { | ||
| // t.Error("profile shouldn't be nil") | ||
| // return | ||
| // } | ||
| // // t.Log(refs) | ||
| // // }(p1, p2) | ||
| // } | ||
| // } | ||
|
|
||
| // wg.Wait() | ||
| } |
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