package net.pterodactylus.fcp.quelaton;
import java.io.IOException;
+import java.util.Objects;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import net.pterodactylus.fcp.AllData;
+import net.pterodactylus.fcp.BaseMessage;
import net.pterodactylus.fcp.CloseConnectionDuplicateClientName;
import net.pterodactylus.fcp.ConfigData;
import net.pterodactylus.fcp.DataFound;
private final Object syncObject = new Object();
private final ListeningExecutorService executorService;
private final FcpConnection fcpConnection;
+ private final Queue<FcpMessage> messages = new ConcurrentLinkedQueue<>();
+ private final AtomicReference<String> identifier = new AtomicReference<>();
public FcpReplySequence(ExecutorService executorService, FcpConnection fcpConnection) {
this.executorService = MoreExecutors.listeningDecorator(executorService);
this.fcpConnection = fcpConnection;
}
+ protected void setIdentifier(String identifier) {
+ this.identifier.set(identifier);
+ }
+
protected abstract boolean isFinished();
public ListenableFuture<R> send(FcpMessage fcpMessage) throws IOException {
- try {
+ setIdentifier(fcpMessage.getField("Identifier"));
fcpConnection.addFcpListener(this);
-
- } catch (Throwable throwable) {
- throwable.printStackTrace();
- }
- fcpConnection.sendMessage(fcpMessage);
+ messages.add(fcpMessage);
return executorService.submit(() -> {
synchronized (syncObject) {
- while (!isFinished()) {
+ while (!isFinished() || !messages.isEmpty()) {
+ while (messages.peek() != null) {
+ FcpMessage message = messages.poll();
+ fcpConnection.sendMessage(message);
+ }
+ if (isFinished()) {
+ continue;
+ }
syncObject.wait();
}
}
});
}
+ protected void sendMessage(FcpMessage fcpMessage) {
+ messages.add(fcpMessage);
+ notifySyncObject();
+ }
+
+ private void notifySyncObject() {
+ synchronized (syncObject) {
+ syncObject.notifyAll();
+ }
+ }
+
protected R getResult() {
return null;
}
fcpConnection.removeFcpListener(this);
}
- private <M> void consume(Consumer<M> consumer, M message) {
- consumer.accept(message);
- synchronized (syncObject) {
- syncObject.notifyAll();
+ private <M extends BaseMessage> void consume(Consumer<M> consumer, M message) {
+ consume(consumer, message, "Identifier");
+ }
+
+ private <M extends BaseMessage> void consume(Consumer<M> consumer, M message,
+ String identifier) {
+ if (Objects.equals(message.getField(identifier), this.identifier.get())) {
+ consumer.accept(message);
+ notifySyncObject();
}
}
private void consumeUnknown(FcpMessage fcpMessage) {
- consumeUnknownMessage(fcpMessage);
- synchronized (syncObject) {
- syncObject.notifyAll();
+ if (Objects.equals(fcpMessage.getField("Identifier"), identifier.get())) {
+ consumeUnknownMessage(fcpMessage);
+ notifySyncObject();
}
}
private void consumeClose(Throwable throwable) {
consumeConnectionClosed(throwable);
- synchronized (syncObject) {
- syncObject.notifyAll();
- }
+ notifySyncObject();
}
@Override
@Override
public final void receivedTestDDAReply(FcpConnection fcpConnection, TestDDAReply testDDAReply) {
- consume(this::consumeTestDDAReply, testDDAReply);
+ consume(this::consumeTestDDAReply, testDDAReply, "Directory");
}
protected void consumeTestDDAReply(TestDDAReply testDDAReply) { }
@Override
public final void receivedTestDDAComplete(FcpConnection fcpConnection, TestDDAComplete testDDAComplete) {
- consume(this::consumeTestDDAComplete, testDDAComplete);
+ consume(this::consumeTestDDAComplete, testDDAComplete, "Directory");
}
protected void consumeTestDDAComplete(TestDDAComplete testDDAComplete) { }