Kaladaran/briar
0
0
Fork 0
mirror of https://code.briarproject.org/briar/briar.git synced 2026-02-14 11:49:04 +01:00
Code Issues Projects Releases Wiki Activity

Replace inner classes with lambdas.

Browse Source
This commit is contained in:
akwizgran
2021-06-10 17:04:15 +01:00
committed by Torsten Grote
parent 9ac72296c7
commit 8be274dc4d
1 changed files with 92 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

240
bramble-core/src/main/java/org/briarproject/bramble/sync/SimplexOutgoingSession.java
View File

@@ -99,9 +99,9 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
// Send our supported protocol versions // Send our supported protocol versions
recordWriter.writeVersions(new Versions(SUPPORTED_VERSIONS)); recordWriter.writeVersions(new Versions(SUPPORTED_VERSIONS));
// Start a query for each type of record // Start a query for each type of record
dbExecutor.execute(new GenerateAck()); dbExecutor.execute(this::generateAck);
if (eager) dbExecutor.execute(new LoadUnackedMessageIds()); if (eager) dbExecutor.execute(this::loadUnackedMessageIds);
else dbExecutor.execute(new GenerateBatch()); else dbExecutor.execute(this::generateBatch);
// Write records until interrupted or no more records to write // Write records until interrupted or no more records to write
try { try {
while (!interrupted) { while (!interrupted) {
@@ -146,166 +146,110 @@ class SimplexOutgoingSession implements SyncSession, EventListener {
} }
} }
private class LoadUnackedMessageIds implements Runnable { @DatabaseExecutor
private void loadUnackedMessageIds() {
@DatabaseExecutor if (interrupted) return;
@Override try {
public void run() { Map<MessageId, Integer> ids = db.transactionWithResult(true, txn ->
if (interrupted) return; db.getUnackedMessagesToSend(txn, contactId));
try { if (LOG.isLoggable(INFO)) {
Map<MessageId, Integer> ids = LOG.info(ids.size() + " unacked messages to send");
db.transactionWithResult(true, txn ->
db.getUnackedMessagesToSend(txn, contactId));
if (LOG.isLoggable(INFO)) {
LOG.info(ids.size() + " unacked messages to send");
}
if (ids.isEmpty()) decrementOutstandingQueries();
else dbExecutor.execute(new GenerateEagerBatch(ids));
} catch (DbException e) {
logException(LOG, WARNING, e);
interrupt();
} }
}
}
private class GenerateEagerBatch implements Runnable {
private final Map<MessageId, Integer> ids;
private GenerateEagerBatch(Map<MessageId, Integer> ids) {
this.ids = ids;
}
@DatabaseExecutor
@Override
public void run() {
if (interrupted) return;
// Take some message IDs from `ids` to form a batch
Collection<MessageId> batchIds = new ArrayList<>();
long totalLength = 0;
Iterator<Entry<MessageId, Integer>> it =
ids.entrySet().iterator();
while (it.hasNext()) {
// Check whether the next message will fit in the batch
Entry<MessageId, Integer> e = it.next();
int length = e.getValue();
if (totalLength + length > MAX_RECORD_PAYLOAD_BYTES) break;
// Add the message to the batch
it.remove();
batchIds.add(e.getKey());
totalLength += length;
}
if (batchIds.isEmpty()) throw new AssertionError();
try {
Collection<Message> batch =
db.transactionWithResult(false, txn ->
db.generateBatch(txn, contactId, batchIds,
maxLatency));
writerTasks.add(new WriteEagerBatch(batch, ids));
} catch (DbException e) {
logException(LOG, WARNING, e);
interrupt();
}
}
}
private class WriteEagerBatch implements ThrowingRunnable<IOException> {
private final Collection<Message> batch;
private final Map<MessageId, Integer> ids;
private WriteEagerBatch(Collection<Message> batch,
Map<MessageId, Integer> ids) {
this.batch = batch;
this.ids = ids;
}
@IoExecutor
@Override
public void run() throws IOException {
if (interrupted) return;
for (Message m : batch) recordWriter.writeMessage(m);
LOG.info("Sent eager batch");
if (ids.isEmpty()) decrementOutstandingQueries(); if (ids.isEmpty()) decrementOutstandingQueries();
else dbExecutor.execute(new GenerateEagerBatch(ids)); else dbExecutor.execute(() -> generateEagerBatch(ids));
} catch (DbException e) {
logException(LOG, WARNING, e);
interrupt();
} }
} }
private class GenerateAck implements Runnable { @DatabaseExecutor
private void generateEagerBatch(Map<MessageId, Integer> ids) {
@DatabaseExecutor if (interrupted) return;
@Override // Take some message IDs from `ids` to form a batch
public void run() { Collection<MessageId> batchIds = new ArrayList<>();
if (interrupted) return; long totalLength = 0;
try { Iterator<Entry<MessageId, Integer>> it = ids.entrySet().iterator();
Ack a = db.transactionWithNullableResult(false, txn -> while (it.hasNext()) {
db.generateAck(txn, contactId, MAX_MESSAGE_IDS)); // Check whether the next message will fit in the batch
if (LOG.isLoggable(INFO)) Entry<MessageId, Integer> e = it.next();
LOG.info("Generated ack: " + (a != null)); int length = e.getValue();
if (a == null) decrementOutstandingQueries(); if (totalLength + length > MAX_RECORD_PAYLOAD_BYTES) break;
else writerTasks.add(new WriteAck(a)); // Add the message to the batch
} catch (DbException e) { it.remove();
logException(LOG, WARNING, e); batchIds.add(e.getKey());
interrupt(); totalLength += length;
} }
if (batchIds.isEmpty()) throw new AssertionError();
try {
Collection<Message> batch =
db.transactionWithResult(false, txn ->
db.generateBatch(txn, contactId, batchIds,
maxLatency));
writerTasks.add(() -> writeEagerBatch(batch, ids));
} catch (DbException e) {
logException(LOG, WARNING, e);
interrupt();
} }
} }
private class WriteAck implements ThrowingRunnable<IOException> { @IoExecutor
private void writeEagerBatch(Collection<Message> batch,
Map<MessageId, Integer> ids) throws IOException {
if (interrupted) return;
for (Message m : batch) recordWriter.writeMessage(m);
LOG.info("Sent eager batch");
if (ids.isEmpty()) decrementOutstandingQueries();
else dbExecutor.execute(() -> generateEagerBatch(ids));
}
private final Ack ack; @DatabaseExecutor
private void generateAck() {
private WriteAck(Ack ack) { if (interrupted) return;
this.ack = ack; try {
} Ack a = db.transactionWithNullableResult(false, txn ->
db.generateAck(txn, contactId, MAX_MESSAGE_IDS));
@IoExecutor if (LOG.isLoggable(INFO))
@Override LOG.info("Generated ack: " + (a != null));
public void run() throws IOException { if (a == null) decrementOutstandingQueries();
if (interrupted) return; else writerTasks.add(() -> writeAck(a));
recordWriter.writeAck(ack); } catch (DbException e) {
LOG.info("Sent ack"); logException(LOG, WARNING, e);
dbExecutor.execute(new GenerateAck()); interrupt();
} }
} }
private class GenerateBatch implements Runnable { @IoExecutor
private void writeAck(Ack ack) throws IOException {
if (interrupted) return;
recordWriter.writeAck(ack);
LOG.info("Sent ack");
dbExecutor.execute(this::generateAck);
}
@DatabaseExecutor @DatabaseExecutor
@Override private void generateBatch() {
public void run() { if (interrupted) return;
if (interrupted) return; try {
try { Collection<Message> b =
Collection<Message> b = db.transactionWithNullableResult(false, txn ->
db.transactionWithNullableResult(false, txn -> db.generateBatch(txn, contactId,
db.generateBatch(txn, contactId, MAX_RECORD_PAYLOAD_BYTES, maxLatency));
MAX_RECORD_PAYLOAD_BYTES, maxLatency)); if (LOG.isLoggable(INFO))
if (LOG.isLoggable(INFO)) LOG.info("Generated batch: " + (b != null));
LOG.info("Generated batch: " + (b != null)); if (b == null) decrementOutstandingQueries();
if (b == null) decrementOutstandingQueries(); else writerTasks.add(() -> writeBatch(b));
else writerTasks.add(new WriteBatch(b)); } catch (DbException e) {
} catch (DbException e) { logException(LOG, WARNING, e);
logException(LOG, WARNING, e); interrupt();
interrupt();
}
} }
} }
private class WriteBatch implements ThrowingRunnable<IOException> { @IoExecutor
private void writeBatch(Collection<Message> batch) throws IOException {
private final Collection<Message> batch; if (interrupted) return;
for (Message m : batch) recordWriter.writeMessage(m);
private WriteBatch(Collection<Message> batch) { LOG.info("Sent batch");
this.batch = batch; dbExecutor.execute(this::generateBatch);
}
@IoExecutor
@Override
public void run() throws IOException {
if (interrupted) return;
for (Message m : batch) recordWriter.writeMessage(m);
LOG.info("Sent batch");
dbExecutor.execute(new GenerateBatch());
}
} }
} }