Rebasing

markets/dealfilter/cli.go

@@ -6,15 +6,39 @@ import (
 	"encoding/json"
 	"os/exec"
 
+	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 
 	"github.com/filecoin-project/lotus/node/modules/dtypes"
 )
 
-func CliDealFilter(cmd string) dtypes.DealFilter {
+func CliStorageDealFilter(cmd string) dtypes.StorageDealFilter {
-	// TODO: run some checks on the cmd string
-
 	return func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error) {
+		d := struct {
+			storagemarket.MinerDeal
+			DealType string
+		}{
+			MinerDeal: deal,
+			DealType:  "storage",
+		}
+		return runDealFilter(ctx, cmd, d)
+	}
+}
+
+func CliRetrievalDealFilter(cmd string) dtypes.RetrievalDealFilter {
+	return func(ctx context.Context, deal retrievalmarket.ProviderDealState) (bool, string, error) {
+		d := struct {
+			retrievalmarket.ProviderDealState
+			DealType string
+		}{
+			ProviderDealState: deal,
+			DealType:          "retrieval",
+		}
+		return runDealFilter(ctx, cmd, d)
+	}
+}
+
+func runDealFilter(ctx context.Context, cmd string, deal interface{}) (bool, string, error) {
 	j, err := json.MarshalIndent(deal, "", "  ")
 	if err != nil {
 		return false, "", err
@@ -35,6 +59,4 @@ func CliDealFilter(cmd string) dtypes.DealFilter {
 	default:
 		return false, "filter cmd run error", err
 	}
-
-	}
 }

node/builder.go

@@ -354,7 +354,8 @@ func Online() Option {
 			Override(new(dtypes.ProviderDataTransfer), modules.NewProviderDAGServiceDataTransfer),
 			Override(new(dtypes.ProviderPieceStore), modules.NewProviderPieceStore),
 			Override(new(*storedask.StoredAsk), modules.NewStorageAsk),
-			Override(new(dtypes.DealFilter), modules.BasicDealFilter(nil)),
+			Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(nil)),
+			Override(new(dtypes.RetrievalDealFilter), modules.RetrievalDealFilter(nil)),
 			Override(new(modules.ProviderDealFunds), modules.NewProviderDealFunds),
 			Override(new(storagemarket.StorageProvider), modules.StorageProvider),
 			Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(nil)),
@@ -484,7 +485,8 @@ func ConfigStorageMiner(c interface{}) Option {
 		ConfigCommon(&cfg.Common),
 
 		If(cfg.Dealmaking.Filter != "",
-			Override(new(dtypes.DealFilter), modules.BasicDealFilter(dealfilter.CliDealFilter(cfg.Dealmaking.Filter))),
+			Override(new(dtypes.StorageDealFilter), modules.BasicDealFilter(dealfilter.CliStorageDealFilter(cfg.Dealmaking.Filter))),
+			Override(new(dtypes.RetrievalDealFilter), modules.RetrievalDealFilter(dealfilter.CliRetrievalDealFilter(cfg.Dealmaking.Filter))),
 		),
 
 		Override(new(storagemarket.StorageProviderNode), storageadapter.NewProviderNodeAdapter(&cfg.Fees)),

node/modules/dtypes/miner.go

@@ -7,6 +7,7 @@ import (
 	"github.com/ipfs/go-cid"
 
 	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
 	"github.com/filecoin-project/go-state-types/abi"
 
@@ -71,4 +72,5 @@ type SetExpectedSealDurationFunc func(time.Duration) error
 // too determine how long sealing is expected to take
 type GetExpectedSealDurationFunc func() (time.Duration, error)
 
-type DealFilter func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error)
+type StorageDealFilter func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error)
+type RetrievalDealFilter func(ctx context.Context, deal retrievalmarket.ProviderDealState) (bool, string, error)

node/modules/storageminer.go

@@ -412,16 +412,16 @@ func NewProviderDealFunds(ds dtypes.MetadataDS) (ProviderDealFunds, error) {
 	return funds.NewDealFunds(ds, datastore.NewKey("/marketfunds/provider"))
 }
 
-func BasicDealFilter(user dtypes.DealFilter) func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
+func BasicDealFilter(user dtypes.StorageDealFilter) func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
 	offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
 	blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
 	expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
-	spn storagemarket.StorageProviderNode) dtypes.DealFilter {
+	spn storagemarket.StorageProviderNode) dtypes.StorageDealFilter {
 	return func(onlineOk dtypes.ConsiderOnlineStorageDealsConfigFunc,
 		offlineOk dtypes.ConsiderOfflineStorageDealsConfigFunc,
 		blocklistFunc dtypes.StorageDealPieceCidBlocklistConfigFunc,
 		expectedSealTimeFunc dtypes.GetExpectedSealDurationFunc,
-		spn storagemarket.StorageProviderNode) dtypes.DealFilter {
+		spn storagemarket.StorageProviderNode) dtypes.StorageDealFilter {
 
 		return func(ctx context.Context, deal storagemarket.MinerDeal) (bool, string, error) {
 			b, err := onlineOk()
@@ -497,7 +497,7 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
 	pieceStore dtypes.ProviderPieceStore,
 	dataTransfer dtypes.ProviderDataTransfer,
 	spn storagemarket.StorageProviderNode,
-	df dtypes.DealFilter,
+	df dtypes.StorageDealFilter,
 	funds ProviderDealFunds,
 ) (storagemarket.StorageProvider, error) {
 	net := smnet.NewFromLibp2pHost(h)
@@ -511,18 +511,11 @@ func StorageProvider(minerAddress dtypes.MinerAddress,
 	return storageimpl.NewProvider(net, namespace.Wrap(ds, datastore.NewKey("/deals/provider")), store, mds, pieceStore, dataTransfer, spn, address.Address(minerAddress), ffiConfig.SealProofType, storedAsk, funds, opt)
 }
 
-// RetrievalProvider creates a new retrieval provider attached to the provider blockstore
+func RetrievalDealFilter(userFilter dtypes.RetrievalDealFilter) func(onlineOk dtypes.ConsiderOnlineRetrievalDealsConfigFunc,
-func RetrievalProvider(h host.Host, miner *storage.Miner, sealer sectorstorage.SectorManager, full lapi.FullNode, ds dtypes.MetadataDS, pieceStore dtypes.ProviderPieceStore, mds dtypes.StagingMultiDstore, dt dtypes.ProviderDataTransfer, onlineOk dtypes.ConsiderOnlineRetrievalDealsConfigFunc, offlineOk dtypes.ConsiderOfflineRetrievalDealsConfigFunc) (retrievalmarket.RetrievalProvider, error) {
+	offlineOk dtypes.ConsiderOfflineRetrievalDealsConfigFunc) dtypes.RetrievalDealFilter {
-	adapter := retrievaladapter.NewRetrievalProviderNode(miner, sealer, full)
+	return func(onlineOk dtypes.ConsiderOnlineRetrievalDealsConfigFunc,
-
+		offlineOk dtypes.ConsiderOfflineRetrievalDealsConfigFunc) dtypes.RetrievalDealFilter {
-	maddr, err := minerAddrFromDS(ds)
+		return func(ctx context.Context, state retrievalmarket.ProviderDealState) (bool, string, error) {
-	if err != nil {
-		return nil, err
-	}
-
-	netwk := rmnet.NewFromLibp2pHost(h)
-
-	opt := retrievalimpl.DealDeciderOpt(func(ctx context.Context, state retrievalmarket.ProviderDealState) (bool, string, error) {
 			b, err := onlineOk()
 			if err != nil {
 				return false, "miner error", err
@@ -542,8 +535,37 @@ func RetrievalProvider(h host.Host, miner *storage.Miner, sealer sectorstorage.S
 				log.Info("offline retrieval has not been implemented yet")
 			}
 
+			if userFilter != nil {
+				return userFilter(ctx, state)
+			}
+
 			return true, "", nil
-	})
+		}
+	}
+}
+
+// RetrievalProvider creates a new retrieval provider attached to the provider blockstore
+func RetrievalProvider(h host.Host,
+	miner *storage.Miner,
+	sealer sectorstorage.SectorManager,
+	full lapi.FullNode,
+	ds dtypes.MetadataDS,
+	pieceStore dtypes.ProviderPieceStore,
+	mds dtypes.StagingMultiDstore,
+	dt dtypes.ProviderDataTransfer,
+	onlineOk dtypes.ConsiderOnlineRetrievalDealsConfigFunc,
+	offlineOk dtypes.ConsiderOfflineRetrievalDealsConfigFunc,
+	userFilter dtypes.RetrievalDealFilter,
+) (retrievalmarket.RetrievalProvider, error) {
+	adapter := retrievaladapter.NewRetrievalProviderNode(miner, sealer, full)
+
+	maddr, err := minerAddrFromDS(ds)
+	if err != nil {
+		return nil, err
+	}
+
+	netwk := rmnet.NewFromLibp2pHost(h)
+	opt := retrievalimpl.DealDeciderOpt(retrievalimpl.DealDecider(userFilter))
 
 	return retrievalimpl.NewProvider(maddr, adapter, netwk, pieceStore, mds, dt, namespace.Wrap(ds, datastore.NewKey("/retrievals/provider")), opt)
 }