remote: make Fetch atomic. (#185)

* Remote now exposes only Fetch. No Connect, Disconnect, etc.
* Repository uses a private fetch method in Remote for Clone/Pull.
* getting capabilities, HEAD or other information from remote
  requires using the lower level client.
* add Fetch method to Repository.

1 files changed, 140 insertions, 139 deletions

diff --git a/remote.go b/remote.go
index 5d1a8c0..f0fb82c 100644
--- a/remote.go
+++ b/remote.go
@@ -25,13 +25,6 @@ type Remote struct {
 	c *config.RemoteConfig
 	s Storer
 	p sideband.Progress
-
-	// cache fields, there during the connection is open
-	endpoint     transport.Endpoint
-	client       transport.Client
-	fetchSession transport.FetchPackSession
-	advRefs      *packp.AdvRefs
-	refs         memory.ReferenceStorage
 }
 
 func newRemote(s Storer, p sideband.Progress, c *config.RemoteConfig) *Remote {
@@ -43,86 +36,93 @@ func (r *Remote) Config() *config.RemoteConfig {
 	return r.c
 }
 
-// Connect with the endpoint
-func (r *Remote) Connect() error {
-	if err := r.initClient(); err != nil {
-		return err
-	}
+func (r *Remote) String() string {
+	fetch := r.c.URL
+	push := r.c.URL
 
-	var err error
-	r.fetchSession, err = r.client.NewFetchPackSession(r.endpoint)
-	if err != nil {
-		return err
-	}
+	return fmt.Sprintf("%s\t%s (fetch)\n%[1]s\t%s (push)", r.c.Name, fetch, push)
+}
 
-	return r.retrieveAdvertisedReferences()
+// Fetch fetches references from the remote to the local repository.
+func (r *Remote) Fetch(o *FetchOptions) error {
+	_, err := r.fetch(o)
+	return err
 }
 
-func (r *Remote) initClient() error {
-	var err error
-	r.endpoint, err = transport.NewEndpoint(r.c.URL)
-	if err != nil {
-		return err
+func (r *Remote) fetch(o *FetchOptions) (refs storer.ReferenceStorer, err error) {
+	if o.RemoteName == "" {
+		o.RemoteName = r.c.Name
 	}
 
-	if r.client != nil {
-		return nil
+	if err := o.Validate(); err != nil {
+		return nil, err
 	}
 
-	r.client, err = client.NewClient(r.endpoint)
+	if len(o.RefSpecs) == 0 {
+		o.RefSpecs = r.c.Fetch
+	}
+
+	s, err := r.newFetchPackSession()
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	return nil
-}
+	defer ioutil.CheckClose(s, &err)
 
-func (r *Remote) retrieveAdvertisedReferences() error {
-	var err error
-	r.advRefs, err = r.fetchSession.AdvertisedReferences()
+	ar, err := s.AdvertisedReferences()
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	r.refs, err = r.advRefs.AllReferences()
-	return err
-}
+	req, err := r.newUploadPackRequest(o, ar)
+	if err != nil {
+		return nil, err
+	}
 
-// AdvertisedReferences returns the git-upload-pack advertised references.
-func (r *Remote) AdvertisedReferences() *packp.AdvRefs {
-	return r.advRefs
-}
+	remoteRefs, err := ar.AllReferences()
+	if err != nil {
+		return nil, err
+	}
 
-// Capabilities returns the remote capabilities
-func (r *Remote) Capabilities() *capability.List {
-	return r.advRefs.Capabilities
-}
+	req.Wants, err = getWants(o.RefSpecs, r.s, remoteRefs)
+	if len(req.Wants) == 0 {
+		return remoteRefs, NoErrAlreadyUpToDate
+	}
 
-// Fetch returns a reader using the request
-func (r *Remote) Fetch(o *FetchOptions) (err error) {
-	if err := o.Validate(); err != nil {
-		return err
+	req.Haves, err = getHaves(r.s)
+	if err != nil {
+		return nil, err
 	}
 
-	if len(o.RefSpecs) == 0 {
-		o.RefSpecs = r.c.Fetch
+	if err := r.fetchPack(o, s, req); err != nil {
+		return nil, err
 	}
 
-	refs, err := r.getWantedReferences(o.RefSpecs)
-	if err != nil {
-		return err
+	if err := r.updateLocalReferenceStorage(o.RefSpecs, remoteRefs); err != nil {
+		return nil, err
 	}
 
-	if len(refs) == 0 {
-		return NoErrAlreadyUpToDate
+	return remoteRefs, err
+}
+
+func (r *Remote) newFetchPackSession() (transport.FetchPackSession, error) {
+	ep, err := transport.NewEndpoint(r.c.URL)
+	if err != nil {
+		return nil, err
 	}
 
-	req, err := r.buildRequest(r.s, o, refs)
+	c, err := client.NewClient(ep)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	reader, err := r.fetchSession.FetchPack(req)
+	return c.NewFetchPackSession(ep)
+}
+
+func (r *Remote) fetchPack(o *FetchOptions, s transport.FetchPackSession,
+	req *packp.UploadPackRequest) (err error) {
+
+	reader, err := s.FetchPack(req)
 	if err != nil {
 		return err
 	}
@@ -134,21 +134,37 @@ func (r *Remote) Fetch(o *FetchOptions) (err error) {
 	}
 
 	if err = r.updateObjectStorage(
-		r.buildSidebandIfSupported(req.Capabilities, reader),
+		buildSidebandIfSupported(req.Capabilities, reader, r.p),
 	); err != nil {
 		return err
 	}
 
-	return r.updateLocalReferenceStorage(o.RefSpecs, refs)
+	return err
 }
 
-func (r *Remote) getWantedReferences(spec []config.RefSpec) ([]*plumbing.Reference, error) {
-	var refs []*plumbing.Reference
-	iter, err := r.References()
+func getHaves(localRefs storer.ReferenceStorer) ([]plumbing.Hash, error) {
+	iter, err := localRefs.IterReferences()
 	if err != nil {
-		return refs, err
+		return nil, err
 	}
 
+	var haves []plumbing.Hash
+	err = iter.ForEach(func(ref *plumbing.Reference) error {
+		if ref.Type() != plumbing.HashReference {
+			return nil
+		}
+
+		haves = append(haves, ref.Hash())
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return haves, nil
+}
+
+func getWants(spec []config.RefSpec, localStorer Storer, remoteRefs storer.ReferenceStorer) ([]plumbing.Hash, error) {
 	wantTags := true
 	for _, s := range spec {
 		if !s.IsWildcard() {
@@ -157,60 +173,82 @@ func (r *Remote) getWantedReferences(spec []config.RefSpec) ([]*plumbing.Referen
 		}
 	}
 
-	return refs, iter.ForEach(func(ref *plumbing.Reference) error {
-		if ref.Type() != plumbing.HashReference {
-			return nil
-		}
+	iter, err := remoteRefs.IterReferences()
+	if err != nil {
+		return nil, err
+	}
 
+	wants := map[plumbing.Hash]bool{}
+	err = iter.ForEach(func(ref *plumbing.Reference) error {
 		if !config.MatchAny(spec, ref.Name()) {
 			if !ref.IsTag() || !wantTags {
 				return nil
 			}
 		}
 
-		_, err := r.s.EncodedObject(plumbing.CommitObject, ref.Hash())
-		if err == plumbing.ErrObjectNotFound {
-			refs = append(refs, ref)
+		if ref.Type() == plumbing.SymbolicReference {
+			ref, err = storer.ResolveReference(remoteRefs, ref.Name())
+			if err != nil {
+				return err
+			}
+		}
+
+		if ref.Type() != plumbing.HashReference {
 			return nil
 		}
 
-		return err
-	})
-}
+		hash := ref.Hash()
+		exists, err := commitExists(localStorer, hash)
+		if err != nil {
+			return err
+		}
 
-func (r *Remote) buildRequest(
-	s storer.ReferenceStorer, o *FetchOptions, refs []*plumbing.Reference,
-) (*packp.UploadPackRequest, error) {
-	req := packp.NewUploadPackRequestFromCapabilities(r.advRefs.Capabilities)
+		if !exists {
+			wants[hash] = true
+		}
 
-	if o.Depth != 0 {
-		req.Depth = packp.DepthCommits(o.Depth)
-		req.Capabilities.Set(capability.Shallow)
+		return nil
+	})
+	if err != nil {
+		return nil, err
 	}
 
-	if r.p == nil && r.advRefs.Capabilities.Supports(capability.NoProgress) {
-		req.Capabilities.Set(capability.NoProgress)
+	var result []plumbing.Hash
+	for h := range wants {
+		result = append(result, h)
 	}
 
-	for _, ref := range refs {
-		req.Wants = append(req.Wants, ref.Hash())
-	}
+	return result, nil
+}
 
-	i, err := s.IterReferences()
-	if err != nil {
-		return nil, err
+func commitExists(s storer.EncodedObjectStorer, h plumbing.Hash) (bool, error) {
+	_, err := s.EncodedObject(plumbing.CommitObject, h)
+	if err == plumbing.ErrObjectNotFound {
+		return false, nil
 	}
 
-	err = i.ForEach(func(ref *plumbing.Reference) error {
-		if ref.Type() != plumbing.HashReference {
-			return nil
+	return true, err
+}
+
+func (r *Remote) newUploadPackRequest(o *FetchOptions,
+	ar *packp.AdvRefs) (*packp.UploadPackRequest, error) {
+
+	req := packp.NewUploadPackRequestFromCapabilities(ar.Capabilities)
+
+	if o.Depth != 0 {
+		req.Depth = packp.DepthCommits(o.Depth)
+		if err := req.Capabilities.Set(capability.Shallow); err != nil {
+			return nil, err
 		}
+	}
 
-		req.Haves = append(req.Haves, ref.Hash())
-		return nil
-	})
+	if r.p == nil && ar.Capabilities.Supports(capability.NoProgress) {
+		if err := req.Capabilities.Set(capability.NoProgress); err != nil {
+			return nil, err
+		}
+	}
 
-	return req, err
+	return req, nil
 }
 
 func (r *Remote) updateObjectStorage(reader io.Reader) error {
@@ -235,7 +273,7 @@ func (r *Remote) updateObjectStorage(reader io.Reader) error {
 	return err
 }
 
-func (r *Remote) buildSidebandIfSupported(l *capability.List, reader io.Reader) io.Reader {
+func buildSidebandIfSupported(l *capability.List, reader io.Reader, p sideband.Progress) io.Reader {
 	var t sideband.Type
 
 	switch {
@@ -248,12 +286,12 @@ func (r *Remote) buildSidebandIfSupported(l *capability.List, reader io.Reader)
 	}
 
 	d := sideband.NewDemuxer(t, reader)
-	d.Progress = r.p
+	d.Progress = p
 
 	return d
 }
 
-func (r *Remote) updateLocalReferenceStorage(specs []config.RefSpec, refs []*plumbing.Reference) error {
+func (r *Remote) updateLocalReferenceStorage(specs []config.RefSpec, refs memory.ReferenceStorage) error {
 	for _, spec := range specs {
 		for _, ref := range refs {
 			if !spec.Match(ref.Name()) {
@@ -272,11 +310,11 @@ func (r *Remote) updateLocalReferenceStorage(specs []config.RefSpec, refs []*plu
 		}
 	}
 
-	return r.buildFetchedTags()
+	return r.buildFetchedTags(refs)
 }
 
-func (r *Remote) buildFetchedTags() error {
-	iter, err := r.References()
+func (r *Remote) buildFetchedTags(refs storer.ReferenceStorer) error {
+	iter, err := refs.IterReferences()
 	if err != nil {
 		return err
 	}
@@ -306,40 +344,3 @@ func (r *Remote) updateShallow(o *FetchOptions, resp *packp.UploadPackResponse)
 
 	return r.s.SetShallow(resp.Shallows)
 }
-
-// Head returns the Reference of the HEAD
-func (r *Remote) Head() *plumbing.Reference {
-	ref, err := storer.ResolveReference(r.refs, plumbing.HEAD)
-	if err != nil {
-		return nil
-	}
-
-	return ref
-}
-
-// Reference returns a Reference for a ReferenceName.
-func (r *Remote) Reference(name plumbing.ReferenceName, resolved bool) (*plumbing.Reference, error) {
-	if resolved {
-		return storer.ResolveReference(r.refs, name)
-	}
-
-	return r.refs.Reference(name)
-}
-
-// References returns an iterator for all references.
-func (r *Remote) References() (storer.ReferenceIter, error) {
-	return r.refs.IterReferences()
-}
-
-// Disconnect from the remote and save the config
-func (r *Remote) Disconnect() error {
-	r.advRefs = nil
-	return r.fetchSession.Close()
-}
-
-func (r *Remote) String() string {
-	fetch := r.c.URL
-	push := r.c.URL
-
-	return fmt.Sprintf("%s\t%s (fetch)\n%[1]s\t%s (push)", r.c.Name, fetch, push)
-}