Merge pull request application-research#480 from application-research…

…/re-use-data-transfer-mgr

re-use data transfer manager

autoretrieve/autoretrieve.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/application-research/estuary/config"
 	"github.com/application-research/estuary/constants"
+	datatransfer "github.com/filecoin-project/go-data-transfer"
 	provider "github.com/filecoin-project/index-provider"
 	"github.com/filecoin-project/index-provider/metadata"
 	"github.com/ipfs/go-cid"
@@ -99,12 +100,13 @@ func findNewCids(db *gorm.DB, lastAdvertisement time.Time) ([]cid.Cid, error) {
 // newIndexProvider creates a new index-provider engine to send announcements to storetheindex
 // this needs to keep running continuously because storetheindex
 // will come to fetch advertisements "when it feels like it"
-func NewAutoretrieveEngine(ctx context.Context, cfg *config.Estuary, db *gorm.DB, libp2pHost host.Host, ds datastore.Batching) (*AutoretrieveEngine, error) {
+func NewAutoretrieveEngine(ctx context.Context, cfg *config.Estuary, db *gorm.DB, libp2pHost host.Host, ds datastore.Batching, dtMgr datatransfer.Manager) (*AutoretrieveEngine, error) {
 	newEngine, err := New(
 		WithHost(libp2pHost), // need to be localhost/estuary
 		WithPublisherKind(DataTransferPublisher),
 		WithDirectAnnounce(cfg.Node.IndexerURL),
 		WithDatastore(ds),
+		WithDataTransfer(dtMgr),
 	)
 	if err != nil {
 		return nil, err

autoretrieve/engine.go

@@ -140,11 +140,7 @@ func (e *AutoretrieveEngine) newPublisher() (legs.Publisher, error) {
 		return nil, nil
 	case DataTransferPublisher:
 		dtOpts := []dtsync.Option{dtsync.Topic(e.pubTopic), dtsync.WithExtraData(e.pubExtraGossipData)}
-		if e.pubDT != nil {
-			return dtsync.NewPublisherFromExisting(e.pubDT, e.h, e.pubTopicName, e.lsys, dtOpts...)
-		}
-		ds := dsn.Wrap(e.ds, datastore.NewKey("/legs/dtsync/pub"))
-		return dtsync.NewPublisher(e.h, ds, e.lsys, e.pubTopicName, dtOpts...)
+		return dtsync.NewPublisherFromExisting(e.pubDT, e.h, e.pubTopicName, e.lsys, dtOpts...)
 	case HttpPublisher:
 		return httpsync.NewPublisher(e.pubHttpListenAddr, e.lsys, e.h.ID(), e.key)
 	default:

go.mod

@@ -4,7 +4,7 @@ go 1.17
 
 require (
 	contrib.go.opencensus.io/exporter/prometheus v0.4.0
-	github.com/application-research/filclient v0.2.0
+	github.com/application-research/filclient v0.3.0
 	github.com/application-research/go-bs-autobatch v0.1.0
 	github.com/cenkalti/backoff/v4 v4.1.3
 	github.com/cheggaaa/pb/v3 v3.0.8
@@ -81,7 +81,7 @@ require (
 	go.opentelemetry.io/otel/exporters/jaeger v1.7.0
 	go.opentelemetry.io/otel/sdk v1.7.0
 	go.opentelemetry.io/otel/trace v1.7.0
-	golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 // indirect
+	golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90
 	golang.org/x/net v0.0.0-20220517181318-183a9ca12b87
 	golang.org/x/sys v0.0.0-20220517195934-5e4e11fc645e
 	golang.org/x/xerrors v0.0.0-20220411194840-2f41105eb62f

go.sum

@@ -130,8 +130,8 @@ github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYU
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/apache/thrift v0.12.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
 github.com/apache/thrift v0.13.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
-github.com/application-research/filclient v0.2.0 h1:bon83EgjUxaSNixoBRM3/TVpmNLzFTUZtGxaUIVxOOo=
-github.com/application-research/filclient v0.2.0/go.mod h1:DQWX5A748TqK1TQdIRVM2vgzW+472Gr0bJ/4TpgTXaY=
+github.com/application-research/filclient v0.3.0 h1:dlUj6Sovt+Ia1BCOAL3RWpRPGTLBp3fJDksxcFkM2cE=
+github.com/application-research/filclient v0.3.0/go.mod h1:DQWX5A748TqK1TQdIRVM2vgzW+472Gr0bJ/4TpgTXaY=
 github.com/application-research/go-bs-autobatch v0.1.0 h1:t8oyrVWFR2sSwLmuzZWWY4PNVabTMPR6nUN4I60MZzM=
 github.com/application-research/go-bs-autobatch v0.1.0/go.mod h1:KhzuDmKnRRWRYkLsxzVCLsA5pYFrT8Yu7FIrN8b9dGw=
 github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
@@ -2570,7 +2570,6 @@ golang.org/x/crypto v0.0.0-20220131195533-30dcbda58838/go.mod h1:IxCIyHEi3zRg3s0
 golang.org/x/crypto v0.0.0-20220210151621-f4118a5b28e2/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.0.0-20220518034528-6f7dac969898/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e h1:T8NU3HyQ8ClP4SEE+KbFlg6n0NhuTsN4MyznaarGsZM=
 golang.org/x/crypto v0.0.0-20220525230936-793ad666bf5e/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 h1:Y/gsMcFOcR+6S6f3YeMKl5g+dZMEWqcz5Czj/GWYbkM=
 golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=

main.go

@@ -711,7 +711,7 @@ func main() {
 
 		// Start autoretrieve if not disabled
 		if !cfg.DisableAutoRetrieve {
-			s.Node.ArEngine, err = autoretrieve.NewAutoretrieveEngine(context.Background(), cfg, s.DB, s.Node.Host, s.Node.Datastore)
+			s.Node.ArEngine, err = autoretrieve.NewAutoretrieveEngine(context.Background(), cfg, s.DB, s.Node.Host, s.Node.Datastore, s.FilClient.GetDtMgr())
 			if err != nil {
 				return err
 			}