RetrieveObjectsWithFallback retrieves objects from the specified endpoints. It tries the endpoints in the order they are provided until one of them returns the object. The endpoint strings must contain a %s placeholder that will be replaced with items from the args slice. The objects are put into th
(endpoints []string, args []string, outCh chan *vt.Object, errCh chan error)
| 55 | // replaced with items from the args slice. The objects are put into the outCh |
| 56 | // as they are retrieved. |
| 57 | func (c *APIClient) RetrieveObjectsWithFallback(endpoints []string, args []string, outCh chan *vt.Object, errCh chan error) error { |
| 58 | |
| 59 | // Make sure outCh and errCh are closed |
| 60 | defer close(outCh) |
| 61 | defer close(errCh) |
| 62 | |
| 63 | h := PQueue{} |
| 64 | heap.Init(&h) |
| 65 | |
| 66 | objCh := make(chan PQueueNode) |
| 67 | getWg := &sync.WaitGroup{} |
| 68 | |
| 69 | // Channel used for limiting the number of parallel goroutines |
| 70 | threads := viper.GetInt("threads") |
| 71 | |
| 72 | if threads == 0 { |
| 73 | panic("RetrieveObjects called with 0 threads") |
| 74 | } |
| 75 | |
| 76 | throttler := make(chan interface{}, threads) |
| 77 | |
| 78 | // Read object IDs from the input channel, launch goroutines to retrieve the |
| 79 | // objects and send them through objCh together with a number indicating |
| 80 | // their order in the input. As goroutines run in parallel the objects can |
| 81 | // be sent out of order to objCh, but the order number is used to reorder |
| 82 | // them. |
| 83 | for order, arg := range args { |
| 84 | getWg.Add(1) |
| 85 | go func(order int, arg string) { |
| 86 | throttler <- nil |
| 87 | var obj *vt.Object |
| 88 | var err error |
| 89 | for _, endpoint := range endpoints { |
| 90 | obj, err = c.GetObject(vt.URL(endpoint, arg)) |
| 91 | if err == nil { |
| 92 | objCh <- PQueueNode{Priority: order, Data: obj} |
| 93 | break |
| 94 | } |
| 95 | if apiErr, ok := err.(vt.Error); ok && apiErr.Code == "NotFoundError" { |
| 96 | // Try the next endpoint |
| 97 | } else { |
| 98 | fmt.Fprintln(os.Stderr, err) |
| 99 | os.Exit(1) |
| 100 | } |
| 101 | } |
| 102 | if err != nil { |
| 103 | objCh <- PQueueNode{Priority: order, Data: err} |
| 104 | } |
| 105 | getWg.Done() |
| 106 | <-throttler |
| 107 | }(order, arg) |
| 108 | } |
| 109 | |
| 110 | outWg := &sync.WaitGroup{} |
| 111 | outWg.Add(1) |
| 112 | |
| 113 | // Read objects from objCh, put them into a priority queue and send them in |
| 114 | // their original order to outCh. |
no test coverage detected