Add provider to ipfs and provide when adding/fetching

License: MIT
Signed-off-by: Michael Avila <davidmichaelavila@gmail.com>

core/commands/dag/dag.go

@@ -138,7 +138,11 @@ into an object of the specified format.
 				return err
 			}
 		}
-		return nil
+
+		return cids.ForEach(func(cid cid.Cid) error {
+			nd.Provider.Provide(cid)
+			return nil
+		})
 	},
 	Type: OutputObject{},
 	Encoders: cmds.EncoderMap{
@@ -180,6 +184,8 @@ format.
 			return err
 		}
 
+		nd.Provider.Provide(obj.Cid())
+
 		var out interface{} = obj
 		if len(rem) > 0 {
 			final, _, err := obj.Resolve(rem)
@@ -219,6 +225,8 @@ var DagResolveCmd = &cmds.Command{
 			return err
 		}
 
+		nd.Provider.Provide(lastCid)
+
 		return cmds.EmitOnce(res, &ResolveOutput{
 			Cid:     lastCid,
 			RemPath: path.Join(rem),

core/commands/get.go

@@ -91,6 +91,8 @@ may also specify the level of compression by specifying '-l=<1-9>'.
 			return err
 		}
 
+		node.Provider.Provide(dn.Cid())
+
 		archive, _ := req.Options[archiveOptionName].(bool)
 		reader, err := uarchive.DagArchive(ctx, dn, p.String(), node.DAG, archive, cmplvl)
 		if err != nil {

core/commands/ls.go

@@ -141,6 +141,8 @@ The JSON output contains type information.
 					Hash:  paths[i],
 					Links: outputLinks,
 				}
+
+				nd.Provider.Provide(dagnode.Cid())
 			}
 
 			return cmds.EmitOnce(res, &LsOutput{output})
@@ -177,6 +179,8 @@ The JSON output contains type information.
 					return err
 				}
 			}
+
+			nd.Provider.Provide(dagnode.Cid())
 		}
 		return nil
 	},

core/commands/refs.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	coreiface "github.com/ipfs/go-ipfs/core/coreapi/interface"
 	"io"
 	"strings"
 
@@ -171,6 +172,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/tar.go

@@ -56,6 +56,9 @@ represent it.
 
 		c := node.Cid()
 
+		nd.Provider.Provide(c)
+
+		// TODO: why is this here?
 		fi.FileName()
 		return cmds.EmitOnce(res, &coreiface.AddEvent{
 			Name: fi.FileName(),

core/core.go

@@ -20,6 +20,7 @@ import (
 	"strings"
 	"time"
 
+	provider "github.com/ipfs/go-ipfs/provider"
 	version "github.com/ipfs/go-ipfs"
 	rp "github.com/ipfs/go-ipfs/exchange/reprovide"
 	filestore "github.com/ipfs/go-ipfs/filestore"
@@ -133,6 +134,7 @@ type IpfsNode struct {
 	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
+	Provider     *provider.Provider  // the value provider system
 	IpnsRepub    *ipnsrp.Republisher
 
 	PubSub   *pubsub.PubSub
@@ -317,6 +319,17 @@ func (n *IpfsNode) startLateOnlineServices(ctx context.Context) error {
 		return err
 	}
 
+	// Provider
+
+	// TODO: Specify strategy in cfg
+	strategy := provider.NewProvideAllStrategy(n.DAG)
+	tracker := provider.NewTracker(n.Repo.Datastore())
+	queue := provider.NewQueue("provider", n.Repo.Datastore())
+	n.Provider = provider.NewProvider(ctx, strategy, tracker, queue, n.Routing)
+	go n.Provider.Run()
+
+	// Reprovider
+
 	var keyProvider rp.KeyChanFunc
 
 	switch cfg.Reprovider.Strategy {

core/coreapi/block.go

@@ -48,6 +48,8 @@ func (api *BlockAPI) Put(ctx context.Context, src io.Reader, opts ...caopts.Bloc
 		return nil, err
 	}
 
+	api.node.Provider.Provide(b.Cid())
+
 	return &BlockStat{path: coreiface.IpldPath(b.Cid()), size: len(data)}, nil
 }
 
@@ -62,6 +64,8 @@ func (api *BlockAPI) Get(ctx context.Context, p coreiface.Path) (io.Reader, erro
 		return nil, err
 	}
 
+	api.node.Provider.Provide(rp.Cid())
+
 	return bytes.NewReader(b.RawData()), nil
 }
 
@@ -114,6 +118,8 @@ func (api *BlockAPI) Stat(ctx context.Context, p coreiface.Path) (coreiface.Bloc
 		return nil, err
 	}
 
+	api.node.Provider.Provide(b.Cid())
+
 	return &BlockStat{
 		path: coreiface.IpldPath(b.Cid()),
 		size: len(b.RawData()),

core/coreapi/coreapi.go

@@ -15,8 +15,7 @@ package coreapi
 
 import (
 	"context"
-
-	core "github.com/ipfs/go-ipfs/core"
+	"github.com/ipfs/go-ipfs/core"
 	coreiface "github.com/ipfs/go-ipfs/core/coreapi/interface"
 
 	ipld "gx/ipfs/QmcKKBwfz6FyQdHR2jsXrrF6XeSBXYL86anmWNewpFpoF5/go-ipld-format"

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.node.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.node.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.node.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.node.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.node.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.node.Provider.Provide(node.Cid())
+
 	return node, nil
 }
 

core/coreapi/unixfs.go

@@ -130,6 +130,11 @@ func (api *UnixfsAPI) Add(ctx context.Context, files files.File, opts ...options
 	if err != nil {
 		return nil, err
 	}
+
+	if !settings.Local {
+		api.node.Provider.Provide(nd.Cid())
+	}
+
 	return coreiface.IpfsPath(nd.Cid()), nil
 }
 
@@ -141,6 +146,8 @@ func (api *UnixfsAPI) Get(ctx context.Context, p coreiface.Path) (coreiface.Unix
 		return nil, err
 	}
 
+	api.node.Provider.Provide(nd.Cid())
+
 	return newUnixfsFile(ctx, ses.dag, nd, "", nil)
 }
 

core/corehttp/gateway_handler.go

@@ -499,6 +499,8 @@ func (i *gatewayHandler) putHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	i.node.Provider.Provide(newcid)
+
 	i.addUserHeaders(w) // ok, _now_ write user's headers.
 	w.Header().Set("IPFS-Hash", newcid.String())
 	http.Redirect(w, r, gopath.Join(ipfsPathPrefix, newcid.String(), newPath), http.StatusCreated)
@@ -576,6 +578,8 @@ func (i *gatewayHandler) deleteHandler(w http.ResponseWriter, r *http.Request) {
 	// Redirect to new path
 	ncid := newnode.Cid()
 
+	i.node.Provider.Provide(ncid)
+
 	i.addUserHeaders(w) // ok, _now_ write user's headers.
 	w.Header().Set("IPFS-Hash", ncid.String())
 	http.Redirect(w, r, gopath.Join(ipfsPathPrefix+ncid.String(), path.Join(components[:len(components)-1])), http.StatusCreated)