CommitAt commits a new serialized BSUP sequence to the journal presuming the previous state conformed to the journal position "at". The entry is written at the next position in the log if possible. Otherwise, a write conflict occurs and an error is returned.
(ctx context.Context, at ID, b []byte)
| 110 | // written at the next position in the log if possible. Otherwise, a write |
| 111 | // conflict occurs and an error is returned. |
| 112 | func (q *Queue) CommitAt(ctx context.Context, at ID, b []byte) error { |
| 113 | uri := q.uri(at + 1) |
| 114 | if err := q.engine.PutIfNotExists(ctx, uri, b); err != nil { |
| 115 | if err != storage.ErrNotSupported { |
| 116 | return err |
| 117 | } |
| 118 | //XXX Here, we need to emulate PutIfNotExists using S3's |
| 119 | // strong ordering guarantees. Currently, this is incorrect |
| 120 | // and can race with multiple writers. See issue #2686. |
| 121 | w, err := q.engine.Put(ctx, uri) |
| 122 | if err != nil { |
| 123 | return err |
| 124 | } |
| 125 | _, err = io.Copy(w, bytes.NewReader(b)) |
| 126 | if err != nil { |
| 127 | w.Close() |
| 128 | return err |
| 129 | } |
| 130 | if err := w.Close(); err != nil { |
| 131 | return err |
| 132 | } |
| 133 | } |
| 134 | return q.writeHead(ctx, at+1) |
| 135 | } |
| 136 | |
| 137 | // NewReader returns a BSUP reader that concatenates the journal files |
| 138 | // in sequence from tail to head. Since BSUP is stored in the journal, |