package org.yamcs.http;

import java.util.ArrayList;
import java.util.List;

import org.yamcs.api.Observer;
import org.yamcs.http.WebSocketServerMessageHandler.Priority;
import org.yamcs.logging.Log;
import org.yamcs.protobuf.Reply;
import org.yamcs.http.WebSocketServerMessageHandler.InternalServerMessage;

import com.google.protobuf.Message;

public class WebSocketObserver implements Observer<Message> {

    private Log log;

    private final TopicContext ctx;

    private int messageCount = 0;
    private final boolean lowPriority;
    private boolean cancelled;
    private boolean completed;
    private Runnable cancelHandler;

    private boolean replied;
    private List<Message> pendingMessages = new ArrayList<>(); // Messages received while not yet replied

    public WebSocketObserver(TopicContext ctx) {
        this.ctx = ctx;
        this.lowPriority = ctx.isLowPriority();
        log = new Log(WebSocketObserver.class);
        log.setContext(ctx.toString());

        log.debug("Subscribe {}", ctx.getTopic().getName());
    }

    void sendReply(Reply reply) {
        synchronized (this) { // Guard 'replied' and 'pendingMessages'
            try {
                sendMessage("reply", reply, Priority.HIGH);
            } finally {
                replied = true;
            }

            pendingMessages.forEach(message -> next(message));
            pendingMessages.clear();
        }
    }

    @Override
    public void next(Message message) {
        synchronized (this) {
            if (!replied) {
                pendingMessages.add(message);
                return;
            }
        }

        messageCount++;

        if (!ctx.nettyContext.channel().isOpen()) {
            ctx.cancel(null);
            return;
        }

        sendMessage(ctx.getTopic().getName(), message, lowPriority ? Priority.LOW : Priority.NORMAL);
    }

    private void sendMessage(String type, Message data, Priority priority) {
        ctx.nettyContext.channel()
                .writeAndFlush(new InternalServerMessage(type, ctx.getId(), messageCount, data, priority));
    }

    void cancelCall(String reason) {
        if (!cancelled) {
            if (reason != null) {
                log.debug("Cancelling {} call ({})", ctx.getTopic().getName(), reason);
            } else {
                log.debug("Cancelling {} call", ctx.getTopic().getName());
            }
            cancelled = true;
            if (cancelHandler != null) {
                cancelHandler.run();
            }
        }
    }

    @Override
    public void completeExceptionally(Throwable t) {
        if (completed) {
            throw new IllegalStateException("Observer already completed");
        }
        completed = true;
    }

    @Override
    public void complete() {
        if (completed) {
            throw new IllegalStateException("Observer already completed");
        }
        completed = true;

        if (cancelled) {
            ctx.requestFuture.complete(null);
            return;
        }
    }

    @Override
    public boolean isCancelled() {
        return cancelled;
    }

    @Override
    public void setCancelHandler(Runnable cancelHandler) {
        this.cancelHandler = cancelHandler;
    }
}

@@ -157,46 +157,91 @@
 ler.
-Priority;%0Aimport org.yamcs.logging.Log
+InternalServerMessage;%0Aimport org.yamcs.http.WebSocketServerMessageHandler.Priority
 ;%0Aim
@@ -259,22 +259,19 @@
 mcs.
-protobuf.Reply
+logging.Log
 ;%0Aim
@@ -289,64 +289,22 @@
 mcs.
-http.WebSocketServerMessageHandler.InternalServerMessage
+protobuf.Reply
 ;%0A%0Ai
@@ -1426,84 +1426,229 @@
     
-@Override%0A    public void next(Message message) %7B%0A        synchronized (this
+/*%0A     * Synchronize because we may get called from different threads and messages%0A     * should be passed in-order to netty, matching messageCount.%0A     */%0A    @Override%0A    public synchronized void next(Message message
 ) %7B%0A
@@ -1651,28 +1651,24 @@
 ) %7B%0A        
-    
 if (!replied
@@ -1667,28 +1667,24 @@
 !replied) %7B%0A
-    
             
@@ -1721,36 +1721,32 @@
             
-    
 return;%0A
             
@@ -1733,30 +1733,16 @@
 return;%0A
-            %7D%0A
         

Fix call seq race when coming from multiple threads