/*
 * Copyright (c) 2014 Arne M. Størksen <arne.storksen@tv2.no>.
 *
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 *
 */

package no.tv2.serf.client;

import java.util.Map;
import java.util.concurrent.Callable;
import org.msgpack.type.Value;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

class ListenerTask implements Callable<Boolean> {

    private final static Logger logger = LoggerFactory.getLogger(ListenerTask.class);
    private final Map<Long, ResponseHandler> handlers;
    private final SerfEndpoint serfEndpoint;
    private boolean closed = false;

    public ListenerTask(Map<Long, ResponseHandler> handlers, SerfEndpoint serfEndpoint) {
        this.handlers = handlers;
        this.serfEndpoint = serfEndpoint;
    }

    @Override
    public Boolean call() throws SerfCommunicationException {
        try {
            while (!serfEndpoint.isClosed() && !closed) {
                //Read header of next message
                Map<String, Value> dstMap = serfEndpoint.readNextMap();
                logger.debug("Reading header: {}", dstMap.toString());
                Long seq = decodeSeq(dstMap);
                String error = decodeError(dstMap);

                //Find handler
                if (handlers.containsKey(seq)) {
                    readResponseAndNotifyHandler(seq, error);
                } else {
                    throw new SerfCommunicationException("No handler for seq=" + seq);
                }
            }
        } catch (SerfCommunicationException ex) {
            if (closed) {
                //Closing the socket will throw exception. This prevents a stacktrace when it is closed properly.
                logger.debug("Socket closed");
            } else {
                throw new SerfCommunicationException(ex);
            }
        }
        //Return true means returns without exceptions
        return true;
    }

    private void readResponseAndNotifyHandler(Long seq, String error) throws SerfCommunicationException {
        Map<String, Value> dstMap;
        ResponseHandler handler = handlers.get(seq);
        if (error == null && handler.hasBody()) {
            dstMap = serfEndpoint.readNextMap();
            logger.debug("Reading body: {}", dstMap.toString());
        } else {
            dstMap = null;
        }
        RawSerfResponse response = new RawSerfResponse(seq, error, dstMap);
        handler.addResponse(response);
    }

    private Long decodeSeq(Map<String, Value> dstMap) {
        Long seq = dstMap.get("Seq").asIntegerValue().getLong();
        return seq;
    }

    private String decodeError(Map<String, Value> dstMap) {
        String error = dstMap.get("Error").asRawValue().getString();
        if (error.length() == 0) {
            error = null;
        }
        return error;
    }

    public void close() {
        this.closed = true;
    }
}

@@ -1713,16 +1713,17 @@
       //
+ 
 Closing 
@@ -1731,81 +1731,146 @@
 he s
-ocket will throw exception. This prevents a stacktrace when it is
+erfEndPoint will cause an exception because we are still in waiting for%0A                // the ReadNextMap call to return.  But if
  closed 
 prop
@@ -1869,17 +1869,34 @@
 sed 
-properly.
+= true, it was intentional
 %0A   
@@ -1952,32 +1952,255 @@
        %7D else %7B%0A
+                // If close is false the socket got closed unexpectedly. Notify the handlers to stop%0A                for (ResponseHandler handler : handlers.values()) %7B%0A                    handler.stop();%0A                %7D%0A
                 
@@ -2319,16 +2319,44 @@
 ceptions
+, BUT nobody cares about it.
 %0A       

Fix streamsubscription no longer receiving events without any form of notification