| 200 | } |
| 201 | |
| 202 | func readID(ctx context.Context, engine storage.Engine, path *storage.URI) (ID, ID, error) { |
| 203 | var retry int |
| 204 | timeout := time.Millisecond |
| 205 | for { |
| 206 | b, err := storage.Get(ctx, engine, path) |
| 207 | if err != nil { |
| 208 | return Nil, Nil, err |
| 209 | } |
| 210 | list := strings.Split(string(b), " ") |
| 211 | if id, err := strconv.ParseUint(list[0], 10, 64); err == nil { |
| 212 | if len(list) == 1 { |
| 213 | return ID(id), Nil, nil |
| 214 | } |
| 215 | if base, err := strconv.ParseUint(list[1], 10, 64); err == nil { |
| 216 | return ID(id), ID(base), nil |
| 217 | } |
| 218 | } |
| 219 | retry++ |
| 220 | if retry > MaxReadRetry || timeout > 5*time.Second { |
| 221 | return Nil, Nil, fmt.Errorf("can read but not parse contents of journal HEAD: %s", b) |
| 222 | } |
| 223 | select { |
| 224 | case <-time.After(timeout): |
| 225 | case <-ctx.Done(): |
| 226 | return Nil, Nil, ctx.Err() |
| 227 | } |
| 228 | t := 2 * int(timeout) |
| 229 | timeout = time.Duration(t + rand.Intn(t)) |
| 230 | } |
| 231 | } |
| 232 | |
| 233 | func Create(ctx context.Context, engine storage.Engine, path *storage.URI, base ID) (*Queue, error) { |
| 234 | q := New(engine, path) |