RFC#2: Add provider to ipfs and provide when adding/fetching (WIP)

core/commands/bitswap.go

@@ -26,7 +26,6 @@ var BitswapCmd = &cmds.Command{
 		"stat":      bitswapStatCmd,
 		"wantlist":  showWantlistCmd,
 		"ledger":    ledgerCmd,
-		"reprovide": reprovideCmd,
 	},
 }
 
@@ -204,29 +203,3 @@ prints the ledger associated with a given peer.
 		}),
 	},
 }
-
-var reprovideCmd = &cmds.Command{
-	Helptext: cmdkit.HelpText{
-		Tagline: "Trigger reprovider.",
-		ShortDescription: `
-Trigger reprovider to announce our data to network.
-`,
-	},
-	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
-		nd, err := cmdenv.GetNode(env)
-		if err != nil {
-			return err
-		}
-
-		if !nd.IsOnline {
-			return ErrNotOnline
-		}
-
-		err = nd.Reprovider.Trigger(req.Context)
-		if err != nil {
-			return err
-		}
-
-		return nil
-	},
-}

core/commands/dag/dag.go

@@ -128,7 +128,26 @@ into an object of the specified format.
 			return err
 		}
 
+//<<<<<<< HEAD
 		return nil
+//=======
+//		if dopin {
+//			cids.ForEach(func(c cid.Cid) error {
+//				nd.Pinning.PinWithMode(c, pin.Recursive)
+//				return nil
+//			})
+//
+//			err := nd.Pinning.Flush()
+//			if err != nil {
+//				return err
+//			}
+//		}
+//
+//		return cids.ForEach(func(cid cid.Cid) error {
+//			nd.Provider.Provide(cid)
+//			return nil
+//		})
+//>>>>>>> Add provider to ipfs and provide when adding/fetching
 	},
 	Type: OutputObject{},
 	Encoders: cmds.EncoderMap{
@@ -175,6 +194,8 @@ format.
 			return err
 		}
 
+		//nd.Provider.Provide(obj.Cid())
+
 		var out interface{} = obj
 		if len(rp.Remainder()) > 0 {
 			rem := strings.Split(rp.Remainder(), "/")
@@ -215,6 +236,8 @@ var DagResolveCmd = &cmds.Command{
 			return err
 		}
 
+		//nd.Provider.Provide(lastCid)
+
 		return cmds.EmitOnce(res, &ResolveOutput{
 			Cid:     rp.Cid(),
 			RemPath: rp.Remainder(),

core/commands/provider.go

@@ -0,0 +1,45 @@
+package commands
+
+import (
+	cmdenv "github.com/ipfs/go-ipfs/core/commands/cmdenv"
+
+	cmds "github.com/ipfs/go-ipfs-cmds"
+	cmdkit "github.com/ipfs/go-ipfs-cmdkit"
+)
+
+var ProviderCmd = &cmds.Command{
+	Helptext: cmdkit.HelpText{
+		Tagline:          "Interact with the provider module.",
+		ShortDescription: ``,
+	},
+
+	Subcommands: map[string]*cmds.Command{
+		"reprovide": reprovideCmd,
+	},
+}
+
+var reprovideCmd = &cmds.Command{
+	Helptext: cmdkit.HelpText{
+		Tagline: "Trigger reprovider.",
+		ShortDescription: `
+Trigger reprovider to announce our data to network.
+`,
+	},
+	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
+		nd, err := cmdenv.GetNode(env)
+		if err != nil {
+			return err
+		}
+
+		if !nd.IsOnline {
+			return ErrNotOnline
+		}
+
+		err = nd.Reprovider.Trigger(req.Context)
+		if err != nil {
+			return err
+		}
+
+		return nil
+	},
+}

core/commands/refs.go

@@ -177,6 +177,9 @@ func objectsForPaths(ctx context.Context, n *core.IpfsNode, paths []string) ([]i
 		if err != nil {
 			return nil, err
 		}
+
+		n.Provider.Provide(o.Cid())
+
 		objects[i] = o
 	}
 	return objects, nil

core/commands/root.go

@@ -57,6 +57,7 @@ ADVANCED COMMANDS
   stats         Various operational stats
   p2p           Libp2p stream mounting
   filestore     Manage the filestore (experimental)
+  provider      Manage the provider system
 
 NETWORK COMMANDS
   id            Show info about IPFS peers
@@ -140,6 +141,7 @@ var rootSubcommands = map[string]*cmds.Command{
 	"pin":       PinCmd,
 	"ping":      PingCmd,
 	"p2p":       P2PCmd,
+	"provider":  ProviderCmd,
 	"refs":      RefsCmd,
 	"resolve":   ResolveCmd,
 	"swarm":     SwarmCmd,

core/commands/tar.go

@@ -43,6 +43,11 @@ represent it.
 			return err
 		}
 
+		api, err := cmdenv.GetApi(env, req)
+		if err != nil {
+			return err
+		}
+
 		enc, err := cmdenv.GetCidEncoder(req)
 		if err != nil {
 			return err
@@ -60,6 +65,7 @@ represent it.
 		}
 
 		c := node.Cid()
+		api.Provider().Provide(c)
 
 		return cmds.EmitOnce(res, &AddEvent{
 			Name: it.Name(),

core/core.go

@@ -21,13 +21,13 @@ import (
 	"time"
 
 	version "github.com/ipfs/go-ipfs"
-	rp "github.com/ipfs/go-ipfs/exchange/reprovide"
 	filestore "github.com/ipfs/go-ipfs/filestore"
 	mount "github.com/ipfs/go-ipfs/fuse/mount"
 	namesys "github.com/ipfs/go-ipfs/namesys"
 	ipnsrp "github.com/ipfs/go-ipfs/namesys/republisher"
 	p2p "github.com/ipfs/go-ipfs/p2p"
 	pin "github.com/ipfs/go-ipfs/pin"
+	provider "github.com/ipfs/go-ipfs/provider"
 	repo "github.com/ipfs/go-ipfs/repo"
 
 	bitswap "github.com/ipfs/go-bitswap"
@@ -124,7 +124,8 @@ type IpfsNode struct {
 	Routing      routing.IpfsRouting // the routing system. recommend ipfs-dht
 	Exchange     exchange.Interface  // the block exchange + strategy (bitswap)
 	Namesys      namesys.NameSystem  // the name system, resolves paths to hashes
-	Reprovider   *rp.Reprovider      // the value reprovider system
+	Reprovider   *provider.Reprovider      // the value reprovider system
+	Provider     *provider.Provider  // the value provider system
 	IpnsRepub    *ipnsrp.Republisher
 
 	AutoNAT  *autonat.AutoNATService
@@ -324,21 +325,24 @@ func (n *IpfsNode) startLateOnlineServices(ctx context.Context) error {
 		return err
 	}
 
-	var keyProvider rp.KeyChanFunc
+	// Provider
 
-	switch cfg.Reprovider.Strategy {
-	case "all":
-		fallthrough
-	case "":
-		keyProvider = rp.NewBlockstoreProvider(n.Blockstore)
-	case "roots":
-		keyProvider = rp.NewPinnedProvider(n.Pinning, n.DAG, true)
-	case "pinned":
-		keyProvider = rp.NewPinnedProvider(n.Pinning, n.DAG, false)
-	default:
-		return fmt.Errorf("unknown reprovider strategy '%s'", cfg.Reprovider.Strategy)
+	// TODO: Specify strategy in cfg
+	strategy := provider.NewProvideAllStrategy(n.DAG)
+	tracker := provider.NewTracker(n.Repo.Datastore())
+	queue, err := provider.NewQueue("provider", ctx, n.Repo.Datastore())
+	if err != nil {
+		return err
+	}
+	n.Provider = provider.NewProvider(ctx, strategy, tracker, queue, n.Blockstore, n.Routing)
+	go n.Provider.Run()
+
+	// Reprovider - new
+
+	rq, err := provider.NewQueue("reprovider", ctx, n.Repo.Datastore())
+	if err != nil {
+		return err
 	}
-	n.Reprovider = rp.NewReprovider(ctx, n.Routing, keyProvider)
 
 	reproviderInterval := kReprovideFrequency
 	if cfg.Reprovider.Interval != "" {
@@ -350,7 +354,8 @@ func (n *IpfsNode) startLateOnlineServices(ctx context.Context) error {
 		reproviderInterval = dur
 	}
 
-	go n.Reprovider.Run(reproviderInterval)
+	n.Reprovider = provider.NewReprovider(ctx, rq, tracker, reproviderInterval, n.Blockstore, n.Routing)
+	n.Reprovider.Run()
 
 	return nil
 }

core/coreapi/block.go

@@ -57,6 +57,8 @@ func (api *BlockAPI) Put(ctx context.Context, src io.Reader, opts ...caopts.Bloc
 		api.pinning.PinWithMode(b.Cid(), pin.Recursive)
 	}
 
+	api.provider.Provide(b.Cid())
+
 	return &BlockStat{path: coreiface.IpldPath(b.Cid()), size: len(data)}, nil
 }
 
@@ -71,6 +73,8 @@ func (api *BlockAPI) Get(ctx context.Context, p coreiface.Path) (io.Reader, erro
 		return nil, err
 	}
 
+	api.provider.Provide(rp.Cid())
+
 	return bytes.NewReader(b.RawData()), nil
 }
 
@@ -123,6 +127,8 @@ func (api *BlockAPI) Stat(ctx context.Context, p coreiface.Path) (coreiface.Bloc
 		return nil, err
 	}
 
+	api.provider.Provide(b.Cid())
+
 	return &BlockStat{
 		path: coreiface.IpldPath(b.Cid()),
 		size: len(b.RawData()),

core/coreapi/coreapi.go

@@ -17,6 +17,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/ipfs/go-ipfs/provider"
 
 	"github.com/ipfs/go-ipfs/core"
 	"github.com/ipfs/go-ipfs/namesys"
@@ -68,6 +69,9 @@ type CoreAPI struct {
 
 	pubSub *pubsub.PubSub
 
+	provider *provider.Provider
+	reprovider *provider.Reprovider
+
 	checkPublishAllowed func() error
 	checkOnline         func(allowOffline bool) error
 
@@ -139,6 +143,10 @@ func (api *CoreAPI) PubSub() coreiface.PubSubAPI {
 	return (*PubSubAPI)(api)
 }
 
+func (api *CoreAPI) Provider() coreiface.ProviderAPI {
+	return (*ProviderAPI)(api)
+}
+
 // WithOptions returns api with global options applied
 func (api *CoreAPI) WithOptions(opts ...options.ApiOption) (coreiface.CoreAPI, error) {
 	settings := api.parentOpts // make sure to copy
@@ -176,6 +184,9 @@ func (api *CoreAPI) WithOptions(opts ...options.ApiOption) (coreiface.CoreAPI, e
 
 		pubSub: n.PubSub,
 
+		provider: n.Provider,
+		reprovider: n.Reprovider,
+
 		nd:         n,
 		parentOpts: settings,
 	}

core/coreapi/object.go

@@ -54,6 +54,9 @@ func (api *ObjectAPI) New(ctx context.Context, opts ...caopts.ObjectNewOption) (
 	if err != nil {
 		return nil, err
 	}
+
+	api.provider.Provide(n.Cid())
+
 	return n, nil
 }
 
@@ -134,6 +137,8 @@ func (api *ObjectAPI) Put(ctx context.Context, src io.Reader, opts ...caopts.Obj
 		}
 	}
 
+	api.provider.Provide(dagnode.Cid())
+
 	return coreiface.IpfsPath(dagnode.Cid()), nil
 }
 
@@ -231,6 +236,8 @@ func (api *ObjectAPI) AddLink(ctx context.Context, base coreiface.Path, name str
 		return nil, err
 	}
 
+	api.provider.Provide(nnode.Cid())
+
 	return coreiface.IpfsPath(nnode.Cid()), nil
 }
 
@@ -257,6 +264,8 @@ func (api *ObjectAPI) RmLink(ctx context.Context, base coreiface.Path, link stri
 		return nil, err
 	}
 
+	api.provider.Provide(nnode.Cid())
+
 	return coreiface.IpfsPath(nnode.Cid()), nil
 }
 
@@ -294,6 +303,8 @@ func (api *ObjectAPI) patchData(ctx context.Context, path coreiface.Path, r io.R
 		return nil, err
 	}
 
+	api.provider.Provide(pbnd.Cid())
+
 	return coreiface.IpfsPath(pbnd.Cid()), nil
 }
 

core/coreapi/path.go

@@ -27,6 +27,9 @@ func (api *CoreAPI) ResolveNode(ctx context.Context, p coreiface.Path) (ipld.Nod
 	if err != nil {
 		return nil, err
 	}
+
+	api.provider.Provide(node.Cid())
+
 	return node, nil
 }
 

core/coreapi/provider.go

@@ -0,0 +1,15 @@
+package coreapi
+
+import (
+	cid "github.com/ipfs/go-cid"
+)
+
+type ProviderAPI CoreAPI
+
+func (api *ProviderAPI) Provide(root cid.Cid) error {
+	return api.provider.Provide(root)
+}
+
+func (api *ProviderAPI) Unprovide(root cid.Cid) error {
+	return api.provider.Unprovide(root)
+}