Class BufferedLedgerClient
- All Implemented Interfaces:
Closeable,AutoCloseable
Callers keep making one-at-a-time calls (transfer, reserve,
commit, rollback, their by-key and tiered forms, and
rejects -- every kind of transfer is batched). Whatever is
outstanding when a connection frees up is sent as a single batch frame, so
the batch size is set by the offered load rather than by a guess -- the
classic self-clocking arrangement, which needs no tuning to be right at both
ends of the load range.
Two properties matter for throughput:
- I/O happens outside the lock. The lock is held only to append to, or swap out, a queue, so producers keep filling the next batch while the current one is on the wire.
- There is a persistent flusher, not a timer per batch.
Each operation type has its own queue, so commits batch with commits and do not queue behind transfers -- a 2PC workload batches on both legs.
Errors. One rule, shared by every TulaDB SDK: a queued item's
failure (its own status, a refusal of its whole frame, a transport error) is
thrown only to the caller who submitted that item. flush() and
close() wait for those frames but never rethrow their errors.
Set pipelineDepth to keep several batch frames in flight per
connection, and pass a LedgerClientPool as the target to get
flushWorkers connections going at once. In-flight transfers are then
bounded by flushWorkers × pipelineDepth × maxBatch. maxBatch
(default and ceiling LedgerClient.TXN_BATCH_MAX) caps the items in
one frame.
Everything else. Built over a single LedgerClient (the
BufferedLedgerClient(LedgerClient) constructors or
of(LedgerClient)), this is a complete client: reads, registry and
admin calls, explicit batches and rejects first wait for what is queued, then
run directly on that connection, so a read never observes less than what a
caller has already been told was committed. Over a pool or a
SmartLedgerClient those calls throw
UnsupportedOperationException; make them on that client.
try (BufferedLedgerClient buffered = new BufferedLedgerClient(new LedgerClient("127.0.0.1", 7700, 30_000))) {
buffered.connect();
TransferResult r = buffered.transfer(transfer); // called from any thread
Balance b = buffered.getBalance(key); // flushes first, then reads
}
LedgerClientPool pool = new LedgerClientPool("ledger://n1:7700,n2:7700", 8, 30_000);
try (BufferedLedgerClient buffered =
new BufferedLedgerClient(BufferedLedgerClient.of(pool), 2.0, 4, 8)) {
TransferResult r = buffered.transfer(transfer); // called from any thread
}
-
Nested Class Summary
Nested ClassesModifier and TypeClassDescriptionstatic interfaceThe batch surface this client drives. -
Constructor Summary
ConstructorsConstructorDescriptionBufferedLedgerClient(BufferedLedgerClient.Target target, double flushIntervalMs, int pipelineDepth, int flushWorkers) BufferedLedgerClient(BufferedLedgerClient.Target target, double flushIntervalMs, int pipelineDepth, int flushWorkers, int maxBatch) BufferedLedgerClient(LedgerClient client) Over one connection, with the defaults (2 ms, one frame in flight, 1024 per frame).BufferedLedgerClient(LedgerClient client, double flushIntervalMs, int maxBatch) Over one connection. -
Method Summary
Modifier and TypeMethodDescriptionclient()The connection this client wraps;nullover a pool or smart client.voidclose()commitBatch(List<String> transferIds) commitByKey(long sourceId, int ledgerId, int currency, String txnId) Commit addressed by its key: the id is derived here, then queued with other commits.voidconnect()Connect the wrapped connection (no-op over a pool or smart client).voidvoidflush()Send everything buffered and wait until every frame is answered.longThe idle flusher's wake-up interval, in nanoseconds.longBatch frames sent so far (including frames that failed).getAccountHistory(Models.AccountKey key, long afterTimestamp, int limit) getBalance(Models.AccountKey key, long minBatchSeq) getCurrency(int numeric) shortR2-H2: read session-wide default flags on the underlying client.getHistory(Models.AccountKey key, long after, int limit) getHistory(Models.AccountKey key, long after, int afterIndex, int limit) getLedger(int ledgerId) getSource(int numeric) getTransaction(String id) getTransactionByKey(long sourceId, int ledgerId, int currency, String txnId) getTransactionByTransferId(byte[] transferId) R2-H4: build-identity probe.booleanWhether the underlying client still holds a socket.longItems sent so far, across all frames.voidledgerAddSource(int ledgerId, int sourceNumeric) ledgerListSources(int ledgerId) voidledgerRemoveSource(int ledgerId, int sourceNumeric) voidledgerSetExpiry(int ledgerId, long reserveTtlNs) voidledgerSetMaxAmount(int ledgerId, long maxTxnAmount) listAccountBalances(int ledgerId, long accountHi, long accountLo) listAccounts(int offset, int limit) listPending(long afterSeq, int limit) listPending(long afterSeq, int afterIndex, int limit) listRejected(long afterSeq, int limit) listRejected(long afterSeq, int afterIndex, int limit) listTransfers(long afterSeq, int limit) listTransfers(long afterSeq, int afterIndex, int limit) listTransfersExtended(int maxRecords, int pageLimit, boolean includePending) listWrappedTransactions(int maxRecords, int pageLimit, boolean includePending) intmaxBatch()The most items this client puts in one frame.static BufferedLedgerClient.Targetof(LedgerClient c) Adapter for a single connection.static BufferedLedgerClient.Targetstatic BufferedLedgerClient.Targetintvoidping()intreject(Models.RejectCreate reject) Record a durable rejected outcome.rejectByKey(long sourceId, int currency, String txnId) rejectByKey(long sourceId, int currency, String txnId, long amount) rejectByKey(long sourceId, int currency, String txnId, long amount, int ledgerId, int reasonCode, Object custom) rejectByKey(long sourceId, int currency, String txnId, long amount, int ledgerId, long debitHi, long debitLo, long creditHi, long creditLo, int reasonCode, Object custom) reserveBatch(List<Models.TransferCreate> transfers) rollbackBatch(List<String> transferIds) rollbackByKey(long sourceId, int ledgerId, int currency, String txnId) searchTransactions(Models.TxnSearch search) voidsetAccountActive(Models.AccountKey key, boolean active) setCurrencyActive(int numeric, boolean active) voidsetDefaultFlags(short flags) R2-H2: set session-wide default flags on the underlying client.setSourceActive(int numeric, boolean active) tieredCommit(String transferId) tieredCommitBatch(List<String> transferIds) tieredCommitByKey(long sourceId, int ledgerId, int currency, String txnId) tieredReserveBatch(List<Models.TieredTransferCreate> transfers) tieredRollback(String transferId) tieredRollbackBatch(List<String> transferIds) tieredRollbackByKey(long sourceId, int ledgerId, int currency, String txnId) tieredTransferBatch(List<Models.TieredTransferCreate> transfers) tieredTransferBatch(List<Models.TieredTransferCreate> transfers, int pipelineDepth) As tieredTransferBatch, keepingpipelineDepthframes in flight.transferBatch(List<Models.TransferCreate> transfers) transferBatch(List<Models.TransferCreate> transfers, int pipelineDepth) As transferBatch, keepingpipelineDepthframes in flight.voidupdateAccountLimitHigh(Models.AccountKey key, long limitHigh) voidupdateAccountLimitLow(Models.AccountKey key, long limitLow)
-
Constructor Details
-
BufferedLedgerClient
Over one connection, with the defaults (2 ms, one frame in flight, 1024 per frame). -
BufferedLedgerClient
Over one connection.- Parameters:
flushIntervalMs- longest the flusher sleeps with nothing to do, in ms; fractions are honoured (0.5 is 500 us)maxBatch- most items in one frame (clamped to 1..TXN_BATCH_MAX)
-
BufferedLedgerClient
-
BufferedLedgerClient
public BufferedLedgerClient(BufferedLedgerClient.Target target, double flushIntervalMs, int pipelineDepth, int flushWorkers) - Parameters:
flushIntervalMs- longest a queued item waits when no wake-up arrivespipelineDepth- batch frames in flight per connectionflushWorkers- concurrent flushers; use the pool size when the target is aLedgerClientPool, since extra flushers on a single-socket client would only queue behind each other on its io lock
-
BufferedLedgerClient
public BufferedLedgerClient(BufferedLedgerClient.Target target, double flushIntervalMs, int pipelineDepth, int flushWorkers, int maxBatch) - Parameters:
maxBatch- most items in one frame (clamped to 1..TXN_BATCH_MAX). Below TXN_BATCH_MAX every slab is one frame, so a pipeline window spans several slabs rather than one.
-
-
Method Details
-
of
Adapter for a single connection. A transport failure or timeout closes the client's connection; the next batch dials again (LedgerClient.reconnect()) instead of every later batch failing with "client is not connected". Once the client is closed by its owner, it stays closed. -
of
-
of
-
pipelineDepth
public int pipelineDepth() -
maxBatch
public int maxBatch()The most items this client puts in one frame. -
flushIntervalNanos
public long flushIntervalNanos()The idle flusher's wake-up interval, in nanoseconds. -
framesSent
public long framesSent()Batch frames sent so far (including frames that failed). -
itemsSent
public long itemsSent()Items sent so far, across all frames. -
pendingCount
public int pendingCount() -
transfer
- Throws:
IOException
-
reserve
- Throws:
IOException
-
commit
- Throws:
IOException
-
rollback
- Throws:
IOException
-
tieredTransfer
- Throws:
IOException
-
tieredReserve
- Throws:
IOException
-
tieredCommit
- Throws:
IOException
-
tieredRollback
- Throws:
IOException
-
commitByKey
public Models.TransferResult commitByKey(long sourceId, int ledgerId, int currency, String txnId) throws IOException Commit addressed by its key: the id is derived here, then queued with other commits.- Throws:
IOException
-
rollbackByKey
public Models.TransferResult rollbackByKey(long sourceId, int ledgerId, int currency, String txnId) throws IOException - Throws:
IOException
-
tieredCommitByKey
public Models.TieredTransferResult tieredCommitByKey(long sourceId, int ledgerId, int currency, String txnId) throws IOException - Throws:
IOException
-
tieredRollbackByKey
public Models.TieredTransferResult tieredRollbackByKey(long sourceId, int ledgerId, int currency, String txnId) throws IOException - Throws:
IOException
-
reject
Record a durable rejected outcome. It rides the transfer queue, so it batches with real traffic, and keeps the single-call mapping ofLedgerClient.reject(com.tuladb.Models.RejectCreate): 41 (recorded) and OK_DUPLICATE succeed; any other item status throws carrying it -- a reject that was not recorded must never read as success.- Throws:
IOException
-
rejectByKey
public Models.TransferResult rejectByKey(long sourceId, int currency, String txnId) throws IOException - Throws:
IOException
-
rejectByKey
public Models.TransferResult rejectByKey(long sourceId, int currency, String txnId, long amount) throws IOException - Throws:
IOException
-
rejectByKey
public Models.TransferResult rejectByKey(long sourceId, int currency, String txnId, long amount, int ledgerId, int reasonCode, Object custom) throws IOException - Throws:
IOException
-
rejectByKey
public Models.TransferResult rejectByKey(long sourceId, int currency, String txnId, long amount, int ledgerId, long debitHi, long debitLo, long creditHi, long creditLo, int reasonCode, Object custom) throws IOException - Throws:
IOException
-
flush
Send everything buffered and wait until every frame is answered.Returns normally once every item queued at call time has been sent and its result delivered. A per-item status or a failed frame is not thrown here: it already reached the caller who submitted the item, exactly once.
- Throws:
IOException- only if the flush cannot complete: the client is closed, or the thread was interrupted while waiting
-
client
The connection this client wraps;nullover a pool or smart client. -
connect
Connect the wrapped connection (no-op over a pool or smart client).- Throws:
IOException
-
ping
- Throws:
IOException
-
getVersion
R2-H4: build-identity probe. Forces a flush so any pending batch lands first, then issues the MSG_VERSION RPC.- Throws:
IOException
-
isConnected
public boolean isConnected()Whether the underlying client still holds a socket. See LedgerClient#isConnected. -
setDefaultFlags
public void setDefaultFlags(short flags) R2-H2: set session-wide default flags on the underlying client. -
getDefaultFlags
public short getDefaultFlags()R2-H2: read session-wide default flags on the underlying client. -
transferBatch
public List<Models.TransferResult> transferBatch(List<Models.TransferCreate> transfers) throws IOException - Throws:
IOException
-
transferBatch
public List<Models.TransferResult> transferBatch(List<Models.TransferCreate> transfers, int pipelineDepth) throws IOException As transferBatch, keepingpipelineDepthframes in flight.- Throws:
IOException
-
reserveBatch
public List<Models.TransferResult> reserveBatch(List<Models.TransferCreate> transfers) throws IOException - Throws:
IOException
-
commitBatch
- Throws:
IOException
-
rollbackBatch
- Throws:
IOException
-
tieredTransferBatch
public List<Models.TieredTransferResult> tieredTransferBatch(List<Models.TieredTransferCreate> transfers, int pipelineDepth) throws IOException As tieredTransferBatch, keepingpipelineDepthframes in flight.- Throws:
IOException
-
tieredTransferBatch
public List<Models.TieredTransferResult> tieredTransferBatch(List<Models.TieredTransferCreate> transfers) throws IOException - Throws:
IOException
-
tieredReserveBatch
public List<Models.TieredTransferResult> tieredReserveBatch(List<Models.TieredTransferCreate> transfers) throws IOException - Throws:
IOException
-
tieredCommitBatch
public List<Models.TieredTransferResult> tieredCommitBatch(List<String> transferIds) throws IOException - Throws:
IOException
-
tieredRollbackBatch
public List<Models.TieredTransferResult> tieredRollbackBatch(List<String> transferIds) throws IOException - Throws:
IOException
-
getBalance
- Throws:
IOException
-
getBalance
- Throws:
IOException
-
listAccountBalances
public List<Models.AccountBalance> listAccountBalances(int ledgerId, long accountHi, long accountLo) throws IOException - Throws:
IOException
-
getTransaction
- Throws:
IOException
-
getTransactionByTransferId
- Throws:
IOException
-
getTransactionByKey
public Models.TxnRecord getTransactionByKey(long sourceId, int ledgerId, int currency, String txnId) throws IOException - Throws:
IOException
-
getHistory
public List<Models.TxnRecord> getHistory(Models.AccountKey key, long after, int limit) throws IOException - Throws:
IOException
-
getHistory
public List<Models.TxnRecord> getHistory(Models.AccountKey key, long after, int afterIndex, int limit) throws IOException - Throws:
IOException
-
getAccountHistory
public List<Models.AccountMetadataEvent> getAccountHistory(Models.AccountKey key, long afterTimestamp, int limit) throws IOException - Throws:
IOException
-
createAccount
- Throws:
IOException
-
createLedger
- Throws:
IOException
-
ledgerSetExpiry
- Throws:
IOException
-
ledgerSetMaxAmount
- Throws:
IOException
-
getLedger
- Throws:
IOException
-
listLedgers
- Throws:
IOException
-
ledgerAddSource
- Throws:
IOException
-
ledgerRemoveSource
- Throws:
IOException
-
ledgerListSources
- Throws:
IOException
-
createSource
- Throws:
IOException
-
getSource
- Throws:
IOException
-
listSources
- Throws:
IOException
-
setSourceActive
- Throws:
IOException
-
setAccountActive
- Throws:
IOException
-
createCurrency
- Throws:
IOException
-
getCurrency
- Throws:
IOException
-
listCurrencies
- Throws:
IOException
-
setCurrencyActive
- Throws:
IOException
-
updateAccountLimitLow
- Throws:
IOException
-
updateAccountLimitHigh
- Throws:
IOException
-
listAccounts
- Throws:
IOException
-
listTransfers
- Throws:
IOException
-
listPending
- Throws:
IOException
-
listRejected
- Throws:
IOException
-
listTransfers
public List<Models.TxnRecord> listTransfers(long afterSeq, int afterIndex, int limit) throws IOException - Throws:
IOException
-
listPending
public List<Models.TxnRecord> listPending(long afterSeq, int afterIndex, int limit) throws IOException - Throws:
IOException
-
listRejected
public List<Models.TxnRecord> listRejected(long afterSeq, int afterIndex, int limit) throws IOException - Throws:
IOException
-
searchTransactions
- Throws:
IOException
-
listTransfersExtended
public List<Models.TxnRecord> listTransfersExtended(int maxRecords, int pageLimit, boolean includePending) throws IOException - Throws:
IOException
-
listWrappedTransactions
public List<LedgerClient.WrappedTxn> listWrappedTransactions(int maxRecords, int pageLimit, boolean includePending) throws IOException - Throws:
IOException
-
close
- Specified by:
closein interfaceAutoCloseable- Specified by:
closein interfaceCloseable- Throws:
IOException
-