melhoria iso-client

gistfile1.txt

Response response = subscribe(response -> response.get(11) == myNsuTerminal).get(30, TimeUnit.SECONDS)


public class ResponseQueue<E> {
    private final ArrayBlockingQueue<E> queue;
    private boolean done;
    
    ResponseQueue() {
        this.queue = new ArrayBlockingQueue<>(1);
        this.done = false;
    }

    public E take(long timeout, TimeUnit unit) {
        return queue.pool(timeout, unit);
    }

    synchronized void putResponse(E response) {
        if (done) {
            throw new IllegalStateException("Response já concluído");
        }
        if (!done) {
            queue.add(response);
            done = true;
        }
    }

    public boolean isDone() {
        return done;
    }
}

private static final subscribes = new ConcurrentHashMap<Predicate, ResponseQueue>();

public ResponseQueue subscribe(Predicate<Response> predicate) {
    if (subscribes.contains(predicate)) {
        throw new IllegalArgumentException("This predicate already exists");
    }
    subscribes.put(predicate, new ResponseQueue<Response>());
}

//Quando uma nova resposta é recebida
public void receivedResponse(Response response) {
    Optional<Entry<Predicate, ResponseQueue>> optTarget = subscribes.entrySet().stream().filter(e -> e.accept(response)).findFirst();
    if (optTarget.isPresent()) {
        //Neste momento o client que estiver esperando a resposta será "notificado" que existe um novo elemento
        optTarget.get().getValue().putResponse(response);
    }
}