-
Notifications
You must be signed in to change notification settings - Fork 58
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
14 changed files
with
212 additions
and
244 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
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,90 @@ | ||
| // Package dtutils provides go-data-transfer related types and functionality for | ||
| // client and provider FSMs | ||
| package dtutils | ||
|
|
||
| import ( | ||
| "errors" | ||
|
|
||
| datatransfer "github.com/filecoin-project/go-data-transfer" | ||
| "github.com/filecoin-project/go-statemachine/fsm" | ||
| logging "github.com/ipfs/go-log/v2" | ||
|
|
||
| "github.com/filecoin-project/go-fil-markets/storagemarket" | ||
| "github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation" | ||
| ) | ||
|
|
||
| var log = logging.Logger("storagemarket_impl") | ||
|
|
||
| var ( | ||
| // ErrDataTransferFailed means a data transfer for a deal failed | ||
| ErrDataTransferFailed = errors.New("deal data transfer failed") | ||
| ) | ||
|
|
||
| // EventReceiver is any thing that can receive FSM events | ||
| type EventReceiver interface { | ||
| Send(id interface{}, name fsm.EventName, args ...interface{}) (err error) | ||
| } | ||
|
|
||
| // DataTransferSubscriber is the function called when an event occurs in a data | ||
| // transfer -- it reads the voucher to verify this even occurred in a storage | ||
| // market deal, then, based on the data transfer event that occurred, it generates | ||
| // and update message for the deal -- either moving to staged for a completion | ||
| // event or moving to error if a data transfer error occurs | ||
| func ProviderDataTransferSubscriber(deals EventReceiver) datatransfer.Subscriber { | ||
| return func(event datatransfer.Event, channelState datatransfer.ChannelState) { | ||
| voucher, ok := channelState.Voucher().(*requestvalidation.StorageDataTransferVoucher) | ||
| // if this event is for a transfer not related to storage, ignore | ||
| if !ok { | ||
| return | ||
| } | ||
|
|
||
| // data transfer events for progress do not affect deal state | ||
| switch event.Code { | ||
| case datatransfer.Open: | ||
| err := deals.Send(voucher.Proposal, storagemarket.ProviderEventDataTransferInitiated) | ||
| if err != nil { | ||
| log.Errorf("processing dt event: %w", err) | ||
| } | ||
| case datatransfer.Complete: | ||
| err := deals.Send(voucher.Proposal, storagemarket.ProviderEventDataTransferCompleted) | ||
| if err != nil { | ||
| log.Errorf("processing dt event: %w", err) | ||
| } | ||
| case datatransfer.Error: | ||
| err := deals.Send(voucher.Proposal, storagemarket.ProviderEventDataTransferFailed, ErrDataTransferFailed) | ||
| if err != nil { | ||
| log.Errorf("processing dt event: %w", err) | ||
| } | ||
| default: | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // DataTransferSubscriber is the function called when an event occurs in a data | ||
| // transfer -- it reads the voucher to verify this even occurred in a storage | ||
| // market deal, then, based on the data transfer event that occurred, it dispatches | ||
| // an event to the appropriate state machine | ||
| func ClientDataTransferSubscriber(deals EventReceiver) datatransfer.Subscriber { | ||
| return func(event datatransfer.Event, channelState datatransfer.ChannelState) { | ||
| voucher, ok := channelState.Voucher().(*requestvalidation.StorageDataTransferVoucher) | ||
| // if this event is for a transfer not related to storage, ignore | ||
| if !ok { | ||
| return | ||
| } | ||
|
|
||
| // data transfer events for progress do not affect deal state | ||
| switch event.Code { | ||
| case datatransfer.Complete: | ||
| err := deals.Send(voucher.Proposal, storagemarket.ClientEventDataTransferComplete) | ||
| if err != nil { | ||
| log.Errorf("processing dt event: %w", err) | ||
| } | ||
| case datatransfer.Error: | ||
| err := deals.Send(voucher.Proposal, storagemarket.ProviderEventDataTransferFailed, ErrDataTransferFailed) | ||
| if err != nil { | ||
| log.Errorf("processing dt event: %w", err) | ||
| } | ||
| default: | ||
| } | ||
| } | ||
| } |
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,103 @@ | ||
| package dtutils_test | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| datatransfer "github.com/filecoin-project/go-data-transfer" | ||
| "github.com/filecoin-project/go-statemachine/fsm" | ||
| "github.com/ipfs/go-cid" | ||
| "github.com/libp2p/go-libp2p-core/peer" | ||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/filecoin-project/go-fil-markets/shared_testutil" | ||
| "github.com/filecoin-project/go-fil-markets/storagemarket" | ||
| "github.com/filecoin-project/go-fil-markets/storagemarket/impl/dtutils" | ||
| "github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation" | ||
| ) | ||
|
|
||
| func TestDataTransferSubscriber(t *testing.T) { | ||
| expectedProposalCID := shared_testutil.GenerateCids(1)[0] | ||
| tests := map[string]struct { | ||
| code datatransfer.EventCode | ||
| called bool | ||
| voucher datatransfer.Voucher | ||
| expectedID interface{} | ||
| expectedEvent fsm.EventName | ||
| expectedArgs []interface{} | ||
| }{ | ||
| "not a storage voucher": { | ||
| called: false, | ||
| voucher: nil, | ||
| }, | ||
| "open event": { | ||
| code: datatransfer.Open, | ||
| called: true, | ||
| voucher: &requestvalidation.StorageDataTransferVoucher{ | ||
| Proposal: expectedProposalCID, | ||
| }, | ||
| expectedID: expectedProposalCID, | ||
| expectedEvent: storagemarket.ProviderEventDataTransferInitiated, | ||
| }, | ||
| "completion event": { | ||
| code: datatransfer.Complete, | ||
| called: true, | ||
| voucher: &requestvalidation.StorageDataTransferVoucher{ | ||
| Proposal: expectedProposalCID, | ||
| }, | ||
| expectedID: expectedProposalCID, | ||
| expectedEvent: storagemarket.ProviderEventDataTransferCompleted, | ||
| }, | ||
| "error event": { | ||
| code: datatransfer.Error, | ||
| called: true, | ||
| voucher: &requestvalidation.StorageDataTransferVoucher{ | ||
| Proposal: expectedProposalCID, | ||
| }, | ||
| expectedID: expectedProposalCID, | ||
| expectedEvent: storagemarket.ProviderEventDataTransferFailed, | ||
| expectedArgs: []interface{}{dtutils.ErrDataTransferFailed}, | ||
| }, | ||
| "other event": { | ||
| code: datatransfer.Progress, | ||
| called: false, | ||
| voucher: &requestvalidation.StorageDataTransferVoucher{ | ||
| Proposal: expectedProposalCID, | ||
| }, | ||
| }, | ||
| } | ||
| for test, data := range tests { | ||
| t.Run(test, func(t *testing.T) { | ||
| fdg := &fakeDealGroup{} | ||
| subscriber := dtutils.ProviderDataTransferSubscriber(fdg) | ||
| subscriber(datatransfer.Event{Code: data.code}, datatransfer.ChannelState{ | ||
| Channel: datatransfer.NewChannel(datatransfer.TransferID(0), cid.Undef, nil, data.voucher, peer.ID(""), peer.ID(""), 0), | ||
| }) | ||
| if data.called { | ||
| require.True(t, fdg.called) | ||
| require.Equal(t, fdg.lastID, data.expectedID) | ||
| require.Equal(t, fdg.lastEvent, data.expectedEvent) | ||
| require.Equal(t, fdg.lastArgs, data.expectedArgs) | ||
| } else { | ||
| require.False(t, fdg.called) | ||
| } | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| // TODO: Test Client DT event subscriber | ||
|
|
||
| type fakeDealGroup struct { | ||
| returnedErr error | ||
| called bool | ||
| lastID interface{} | ||
| lastEvent fsm.EventName | ||
| lastArgs []interface{} | ||
| } | ||
|
|
||
| func (fdg *fakeDealGroup) Send(id interface{}, name fsm.EventName, args ...interface{}) (err error) { | ||
| fdg.lastID = id | ||
| fdg.lastEvent = name | ||
| fdg.lastArgs = args | ||
| fdg.called = true | ||
| return fdg.returnedErr | ||
| } |
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
Oops, something went wrong.