forked from gazette/core
-
Notifications
You must be signed in to change notification settings - Fork 2
/
appender.go
173 lines (151 loc) · 4.86 KB
/
appender.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
package client
import (
"context"
"errors"
"io"
"math"
"time"
pb "github.com/LiveRamp/gazette/v2/pkg/protocol"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// Appender adapts an Append RPC to the io.WriteCloser interface. Its usages
// should be limited to cases where the full and complete buffer to append is
// already available and can be immediately dispatched as, by design, an in-
// progress RPC prevents the broker from serving other Append RPCs concurrently.
type Appender struct {
Request pb.AppendRequest // AppendRequest of the Append.
Response pb.AppendResponse // AppendResponse sent by broker.
ctx context.Context
client pb.RoutedJournalClient // Client against which Read is dispatched.
stream pb.Journal_AppendClient // Server stream.
}
// NewAppender returns an Appender initialized with the BrokerClient and AppendRequest.
func NewAppender(ctx context.Context, client pb.RoutedJournalClient, req pb.AppendRequest) *Appender {
var a = &Appender{
Request: req,
ctx: ctx,
client: client,
}
return a
}
// Reset the Appender to its post-construction state,
// allowing it to be re-used or re-tried.
func (a *Appender) Reset() { a.Response, a.stream = pb.AppendResponse{}, nil }
func (a *Appender) Write(p []byte) (n int, err error) {
if len(p) == 0 {
return // The broker interprets empty chunks as "commit".
}
// Lazy initialization: begin the Append RPC.
if err = a.lazyInit(); err != nil {
// Pass.
} else if err = a.sendMsg(&pb.AppendRequest{Content: p}); err != nil {
// Pass.
} else {
n = len(p)
}
if err != nil {
err = mapGRPCCtxErr(a.ctx, err)
}
return
}
// Close the Append to complete the transaction, committing previously
// written content. If Close returns without an error, Append.Response
// will hold the broker response.
func (a *Appender) Close() (err error) {
// Send an empty chunk to signal commit of previously written content
if err = a.lazyInit(); err != nil {
// Pass.
} else if err = a.sendMsg(new(pb.AppendRequest)); err != nil {
// Pass.
} else if _ = a.stream.CloseSend(); false {
// Ignore CloseSend's error. Currently, gRPC will never return one. If the
// stream is broken, it *could* return io.EOF but we'd rather read the actual
// casual error with RecvMsg.
} else if err = a.stream.RecvMsg(&a.Response); err != nil {
// Pass.
} else if err = a.Response.Validate(); err != nil {
// Pass.
} else {
a.client.UpdateRoute(a.Request.Journal.String(), &a.Response.Header.Route)
switch a.Response.Status {
case pb.Status_OK:
// Pass.
case pb.Status_NOT_JOURNAL_PRIMARY_BROKER:
err = ErrNotJournalPrimaryBroker
case pb.Status_WRONG_APPEND_OFFSET:
err = ErrWrongAppendOffset
default:
err = errors.New(a.Response.Status.String())
}
}
if err != nil {
err = mapGRPCCtxErr(a.ctx, err)
}
return
}
// Abort the write, causing the broker to discard previously written content.
func (a *Appender) Abort() {
if a.stream != nil {
// Abort is implied by sending EOF without a preceding empty chunk.
_, _ = a.stream.CloseAndRecv()
}
}
func (a *Appender) lazyInit() (err error) {
if a.stream == nil {
if a.Request.Journal == "" {
return pb.NewValidationError("expected Request.Journal")
} else if err = a.Request.Validate(); err != nil {
return pb.ExtendContext(err, "Request")
}
a.stream, err = a.client.Append(
pb.WithDispatchItemRoute(a.ctx, a.client, a.Request.Journal.String(), true))
if err == nil {
// Send request preamble metadata prior to append content chunks.
err = a.sendMsg(&a.Request)
}
}
return
}
func (a *Appender) sendMsg(r *pb.AppendRequest) (err error) {
if err = a.stream.SendMsg(r); err == io.EOF {
// EOF indicates that a server-side error has occurred, but it must
// still be read via RecvMsg. See SendMsg docs.
err = a.stream.RecvMsg(&a.Response)
}
return
}
// Append zero or more ReaderAts of |content| to a journal as a single Append
// transaction. Append retries on transport or routing errors, but fails
// on all other errors. If no ReaderAts are provided, an Append RPC with no
// content is issued.
func Append(ctx context.Context, rjc pb.RoutedJournalClient, req pb.AppendRequest,
content ...io.ReaderAt) (pb.AppendResponse, error) {
for attempt := 0; true; attempt++ {
var a = NewAppender(ctx, rjc, req)
var err error
for r := 0; r != len(content) && err == nil; r++ {
_, err = io.Copy(a, io.NewSectionReader(content[r], 0, math.MaxInt64))
}
if err == nil {
err = a.Close()
} else {
a.Abort()
}
if err == nil {
return a.Response, nil
} else if s, ok := status.FromError(err); ok && s.Code() == codes.Unavailable {
// Fallthrough to retry
} else if err == ErrNotJournalPrimaryBroker {
// Fallthrough.
} else {
return a.Response, err
}
select {
case <-ctx.Done():
return a.Response, ctx.Err()
case <-time.After(backoff(attempt)):
}
}
panic("not reached")
}