package io.vertx.ext.stomp;
import io.vertx.core.Handler;
import io.vertx.ext.stomp.impl.Transaction;
import io.vertx.ext.stomp.impl.Transactions;
import io.vertx.ext.stomp.utils.Headers;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
public class DefaultCommitHandler implements Handler<ServerFrame> {
@Override
public void handle(ServerFrame serverFrame) {
Frame frame = serverFrame.frame();
StompServerConnection connection = serverFrame.connection();
String txId = frame.getHeader(Frame.TRANSACTION);
if (txId == null) {
Frame error = Frames.createErrorFrame("Missing transaction id", Headers.create(), "COMMIT frames " +
"must contain the " +
"'transaction' header.");
connection.write(error).close();
return;
}
Transaction transaction = Transactions.instance().getTransaction(connection, txId);
if (transaction == null) {
Frame error = Frames.createErrorFrame("Unknown transaction",
Headers.create(Frame.TRANSACTION, txId),
"The transaction id is unknown.");
connection.write(error).close();
return;
}
replay(connection, transaction.getFrames());
transaction.clear();
Transactions.instance().unregisterTransaction(connection, txId);
Frames.handleReceipt(frame, connection);
}
private void replay(StompServerConnection connection, List<Frame> frames) {
Iterator<Frame> iterator = frames.iterator();
while (iterator.hasNext()) {
List<Frame> chunk = new ArrayList<>();
while (iterator.hasNext() && chunk.size() < connection.server().options().getTransactionChunkSize()) {
chunk.add(iterator.next());
}
connection.server().vertx().runOnContext(v -> replayChunk(connection, chunk));
}
}
private void replayChunk(StompServerConnection connection, List<Frame> frames) {
final List<Destination> destinations = connection.handler().getDestinations();
for (Frame frame : frames) {
switch (frame.getCommand()) {
case SEND:
String destination = frame.getHeader(Frame.DESTINATION);
Destination dest = connection.handler().getDestination(destination);
if (dest != null) {
dest.dispatch(connection, frame);
}
break;
case ACK:
for (Destination d : destinations) {
if (d.ack(connection, frame)) {
break;
}
}
break;
case NACK:
for (Destination d : destinations) {
if (d.nack(connection, frame)) {
break;
}
}
break;
}
}
}
}