5824: Add some clarifying comments and golint/vet/fmt fixes.

[arvados.git] / sdk / go / keepclient / support.go

diff --git a/sdk/go/keepclient/support.go b/sdk/go/keepclient/support.go
index ee308dd0d4a1895d1b4fd6cbcb378cd3a19b4916..f5554618feece00d04c34982347ffcc8e1e405aa 100644 (file)
--- a/sdk/go/keepclient/support.go
+++ b/sdk/go/keepclient/support.go
@@ -2,6 +2,7 @@ package keepclient
 
 import (
        "crypto/md5"
+       "encoding/json"
        "errors"
        "fmt"
        "git.curoverse.com/arvados.git/sdk/go/streamer"
@@ -76,23 +77,38 @@ func (this *KeepClient) setClientSettingsDisk() {
        }
 }
 
+type svcList struct {
+       Items []keepService `json:"items"`
+}
+
 // DiscoverKeepServers gets list of available keep services from api server
 func (this *KeepClient) DiscoverKeepServers() error {
-       type svcList struct {
-               Items []keepService `json:"items"`
-       }
-       var m svcList
+       var list svcList
 
        // Get keep services from api server
-       err := this.Arvados.Call("GET", "keep_services", "", "accessible", nil, &m)
-
-       // If there is error getting keep services, get list of keep disks
+       err := this.Arvados.Call("GET", "keep_services", "", "accessible", nil, &list)
        if err != nil {
-               if err := this.Arvados.List("keep_disks", nil, &m); err != nil {
-                       return err
-               }
+               return err
        }
 
+       return this.loadKeepServers(list)
+}
+
+// LoadKeepServicesFromJSON gets list of available keep services from given JSON
+func (this *KeepClient) LoadKeepServicesFromJSON(services string) error {
+       var list svcList
+
+       // Load keep services from given json
+       dec := json.NewDecoder(strings.NewReader(services))
+       if err := dec.Decode(&list); err != nil {
+               return err
+       }
+
+       return this.loadKeepServers(list)
+}
+
+// loadKeepServers
+func (this *KeepClient) loadKeepServers(list svcList) error {
        listed := make(map[string]bool)
        localRoots := make(map[string]string)
        gatewayRoots := make(map[string]string)
@@ -102,7 +118,7 @@ func (this *KeepClient) DiscoverKeepServers() error {
        this.replicasPerService = 1
        this.Using_proxy = false
 
-       for _, service := range m.Items {
+       for _, service := range list.Items {
                scheme := "http"
                if service.SSL {
                        scheme = "https"
@@ -153,7 +169,7 @@ type uploadStatus struct {
        response        string
 }
 
-func (this KeepClient) uploadToKeepServer(host string, hash string, body io.ReadCloser,
+func (this *KeepClient) uploadToKeepServer(host string, hash string, body io.ReadCloser,
        upload_status chan<- uploadStatus, expectedLength int64, requestId string) {
 
        var req *http.Request
@@ -198,7 +214,7 @@ func (this KeepClient) uploadToKeepServer(host string, hash string, body io.Read
        defer resp.Body.Close()
        defer io.Copy(ioutil.Discard, resp.Body)
 
-       respbody, err2 := ioutil.ReadAll(&io.LimitedReader{resp.Body, 4096})
+       respbody, err2 := ioutil.ReadAll(&io.LimitedReader{R: resp.Body, N: 4096})
        response := strings.TrimSpace(string(respbody))
        if err2 != nil && err2 != io.EOF {
                log.Printf("[%v] Upload %v error: %v response: %v", requestId, url, err2.Error(), response)
@@ -212,7 +228,7 @@ func (this KeepClient) uploadToKeepServer(host string, hash string, body io.Read
        }
 }
 
-func (this KeepClient) putReplicas(
+func (this *KeepClient) putReplicas(
        hash string,
        tr *streamer.AsyncStream,
        expectedLength int64) (locator string, replicas int, err error) {
@@ -232,7 +248,17 @@ func (this KeepClient) putReplicas(
 
        // Used to communicate status from the upload goroutines
        upload_status := make(chan uploadStatus)
-       defer close(upload_status)
+       defer func() {
+               // Wait for any abandoned uploads (e.g., we started
+               // two uploads and the first replied with replicas=2)
+               // to finish before closing the status channel.
+               go func() {
+                       for active > 0 {
+                               <-upload_status
+                       }
+                       close(upload_status)
+               }()
+       }()
 
        // Desired number of replicas
        remaining_replicas := this.Want_replicas