/
WriteHandle.java
160 lines (146 loc) · 5.75 KB
/
WriteHandle.java
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
/**
*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
*/
package org.apache.bookkeeper.client.api;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.common.annotation.InterfaceAudience.Public;
import org.apache.bookkeeper.common.annotation.InterfaceStability.Unstable;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
/**
* Provide write access to a ledger.
*
* @see WriteAdvHandle
*
* @since 4.6
*/
@Public
@Unstable
public interface WriteHandle extends ReadHandle, ForceableHandle {
/**
* Add entry asynchronously to an open ledger.
*
* @param data a bytebuf to be written. The bytebuf's reference count will be decremented by 1 after the
* completable future is returned
* @return an handle to the result, in case of success it will return the id of the newly appended entry
*/
CompletableFuture<Long> appendAsync(ByteBuf data);
/**
* Add entry synchronously to an open ledger.
*
* @param data a bytebuf to be written. The bytebuf's reference count will be decremented by 1 after the
* call completes.
* @return the id of the newly appended entry
*/
default long append(ByteBuf data) throws BKException, InterruptedException {
return FutureUtils.<Long, BKException>result(appendAsync(data), BKException.HANDLER);
}
/**
* Add entry asynchronously to an open ledger.
*
* @param data array of bytes to be written
* @return an handle to the result, in case of success it will return the id of the newly appended entry
*/
default CompletableFuture<Long> appendAsync(ByteBuffer data) {
return appendAsync(Unpooled.wrappedBuffer(data));
}
/**
* Add entry synchronously to an open ledger.
*
* @param data array of bytes to be written
* @return the id of the newly appended entry
*/
default long append(ByteBuffer data) throws BKException, InterruptedException {
return append(Unpooled.wrappedBuffer(data));
}
/**
* Add an entry asynchronously to an open ledger.
*
* @param data array of bytes to be written
* @return a completable future represents the add result, in case of success the future returns the entry id
* of this newly appended entry
*/
default CompletableFuture<Long> appendAsync(byte[] data) {
return appendAsync(Unpooled.wrappedBuffer(data));
}
/**
* Add an entry synchronously to an open ledger.
*
* @param data array of bytes to be written
* @return the entry id of this newly appended entry
*/
default long append(byte[] data) throws BKException, InterruptedException {
return append(Unpooled.wrappedBuffer(data));
}
/**
* Add an entry asynchronously to an open ledger.
*
* @param data array of bytes to be written
* @param offset the offset in the bytes array
* @param length the length of the bytes to be appended
* @return a completable future represents the add result, in case of success the future returns the entry id
* of this newly appended entry
*/
default CompletableFuture<Long> appendAsync(byte[] data, int offset, int length) {
return appendAsync(Unpooled.wrappedBuffer(data, offset, length));
}
/**
* Add an entry synchronously to an open ledger.
*
* @param data array of bytes to be written
* @param offset the offset in the bytes array
* @param length the length of the bytes to be appended
* @return the entry id of this newly appended entry
*/
default long append(byte[] data, int offset, int length) throws BKException, InterruptedException {
return append(Unpooled.wrappedBuffer(data, offset, length));
}
/**
* Get the entry id of the last entry that has been enqueued for addition (but
* may not have possibly been persisted to the ledger).
*
* @return the entry id of the last entry pushed or -1 if no entry has been pushed
*/
long getLastAddPushed();
/**
* Asynchronous close the write handle, any adds in flight will return errors.
*
* <p>Closing a ledger will ensure that all clients agree on what the last
* entry of the ledger is. Once the ledger has been closed, all reads from the
* ledger will return the same set of entries.
*
* @return an handle to access the result of the operation
*/
@Override
CompletableFuture<Void> closeAsync();
/**
* Synchronous close the write handle, any adds in flight will return errors.
*
* <p>Closing a ledger will ensure that all clients agree on what the last
* entry of the ledger is. Once the ledger has been closed, all reads from the
* ledger will return the same set of entries.
*/
@Override
default void close() throws BKException, InterruptedException {
FutureUtils.<Void, BKException>result(closeAsync(), BKException.HANDLER);
}
}