@@ -16,6 +16,7 @@ package counter
16
16
17
17
import (
18
18
"context"
19
+
19
20
api "github.com/atomix/api/proto/atomix/counter"
20
21
"github.com/atomix/api/proto/atomix/headers"
21
22
"github.com/atomix/go-client/pkg/client/primitive"
@@ -47,6 +48,9 @@ type Counter interface {
47
48
48
49
// Decrement decrements the counter by the given delta
49
50
Decrement (ctx context.Context , delta int64 ) (int64 , error )
51
+
52
+ // CAS check the counter value and then updates its current value
53
+ CAS (ctx context.Context , expect int64 , update int64 ) (bool , error )
50
54
}
51
55
52
56
// New creates a new counter for the given partitions
@@ -149,6 +153,26 @@ func (c *counter) Decrement(ctx context.Context, delta int64) (int64, error) {
149
153
return response .(* api.DecrementResponse ).NextValue , nil
150
154
}
151
155
156
+ func (c * counter ) CAS (ctx context.Context , expect int64 , update int64 ) (bool , error ) {
157
+ response , err := c .instance .DoCommand (ctx , func (ctx context.Context , conn * grpc.ClientConn , header * headers.RequestHeader ) (* headers.ResponseHeader , interface {}, error ) {
158
+ client := api .NewCounterServiceClient (conn )
159
+ request := & api.CheckAndSetRequest {
160
+ Header : header ,
161
+ Expect : expect ,
162
+ Update : update ,
163
+ }
164
+ response , err := client .CheckAndSet (ctx , request )
165
+ if err != nil {
166
+ return nil , nil , err
167
+ }
168
+ return response .Header , response , nil
169
+ })
170
+ if err != nil {
171
+ return false , err
172
+ }
173
+ return response .(* api.CheckAndSetResponse ).Succeeded , nil
174
+ }
175
+
152
176
func (c * counter ) Close (ctx context.Context ) error {
153
177
return c .instance .Close (ctx )
154
178
}
0 commit comments