/*
 *   Copyright 2013 - 2019 The Original Authors
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 */

package org.elasticsoftware.elasticactors.cluster;

import com.google.common.collect.ImmutableList;
import org.elasticsoftware.elasticactors.ActorContainer;
import org.elasticsoftware.elasticactors.ActorContainerRef;
import org.elasticsoftware.elasticactors.ActorRef;
import org.elasticsoftware.elasticactors.PhysicalNode;
import org.elasticsoftware.elasticactors.messaging.InternalMessage;
import org.elasticsoftware.elasticactors.messaging.MessageHandler;
import org.elasticsoftware.elasticactors.messaging.MessageHandlerEventListener;
import org.elasticsoftware.elasticactors.messaging.MessageQueue;
import org.elasticsoftware.elasticactors.messaging.MessageQueueFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Joost van de Wijgerd
 */
public abstract class AbstractActorContainer implements ActorContainer, MessageHandler {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final ActorRef myRef;
    private final MessageQueueFactory messageQueueFactory;
    protected MessageQueue messageQueue;
    private final PhysicalNode localNode;

    public AbstractActorContainer(MessageQueueFactory messageQueueFactory, ActorRef myRef, PhysicalNode node) {
        this.messageQueueFactory = messageQueueFactory;
        this.myRef = myRef;
        this.localNode = node;
    }

    @Override
    public void init() throws Exception {
        this.messageQueue = messageQueueFactory.create(myRef.getActorPath(), this);
    }

    @Override
    public void destroy() {
        // release all resources
        this.messageQueue.destroy();
    }

    @Override
    public final ActorRef getActorRef() {
        return myRef;
    }

    @Override
    public final void offerInternalMessage(InternalMessage message) {
        messageQueue.add(message);
    }

    @Override
    public final PhysicalNode getPhysicalNode() {
        return localNode;
    }

    protected void handleUndeliverable(InternalMessage internalMessage, ActorRef receiverRef, MessageHandlerEventListener messageHandlerEventListener) throws Exception {
        // if a message-undeliverable is undeliverable, don't send an undeliverable message back!
        ActorRef senderRef = internalMessage.getSender();
        try {
            if (senderRef instanceof ActorContainerRef && !internalMessage.isUndeliverable()) {
                ((ActorContainerRef) senderRef).getActorContainer().undeliverableMessage(internalMessage, receiverRef);
            } else if(internalMessage.isUndeliverable()) {
                logger.error("Receiver for undeliverable message not found: message type '{}' , receiver '{}'", internalMessage.getPayloadClass(), receiverRef);
            } else {
                logger.warn("Could not send message undeliverable: original message type '{}' , receiver '{}'", internalMessage.getPayloadClass(), receiverRef);
            }
        } finally {
            // ack anyway
            messageHandlerEventListener.onDone(internalMessage);
        }
    }

    @Override
    public void sendMessage(ActorRef from, ActorRef to, Object message) throws Exception {
        sendMessage(from, ImmutableList.of(to), message);
    }
}

@@ -3122,16 +3122,17 @@
  else if
+ 
 (interna
@@ -3190,16 +3190,41 @@
 r.error(
+%0A                        
 %22Receive
@@ -3264,11 +3264,48 @@
 ound
-: m
+. %22%0A                                + %22M
 essa
@@ -3316,72 +3316,242 @@
 ype 
-'%7B%7D' , receiver '%7B%7D'%22, internalMessage.getPayloadClass(), receiv
+%5B%7B%7D%5D. %22%0A                                + %22Receiver %5B%7B%7D%5D. %22%0A                                + %22Sender %5B%7B%7D%5D.%22,%0A                        internalMessage.getPayloadClass(),%0A                        receiverRef,%0A                        send
 erRe
@@ -3603,16 +3603,41 @@
 er.warn(
+%0A                        
 %22Could n
@@ -3669,11 +3669,48 @@
 able
-: o
+. %22%0A                                + %22O
 rigi
@@ -3730,24 +3730,60 @@
 ype 
-'%7B%7D' , r
+%5B%7B%7D%5D. %22%0A                                + %22R
 eceiver 
 '%7B%7D'
@@ -3782,56 +3782,190 @@
 ver 
-'%7B%7D'%22, internalMessage.getPayloadClass(), receiv
+%5B%7B%7D%5D. %22%0A                                + %22Sender %5B%7B%7D%5D.%22,%0A                        internalMessage.getPayloadClass(),%0A                        receiverRef,%0A                        send
 erRe

Improve logging for undeliverable messages