/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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.apache.qpid.protonj2.client.util;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

import org.apache.qpid.protonj2.buffer.ProtonBuffer;
import org.apache.qpid.protonj2.buffer.ProtonBufferAllocator;
import org.apache.qpid.protonj2.client.AdvancedMessage;
import org.apache.qpid.protonj2.client.Message;
import org.apache.qpid.protonj2.client.exceptions.ClientException;
import org.apache.qpid.protonj2.client.impl.ClientMessageSupport;
import org.apache.qpid.protonj2.types.Binary;
import org.apache.qpid.protonj2.types.Symbol;
import org.apache.qpid.protonj2.types.messaging.ApplicationProperties;
import org.apache.qpid.protonj2.types.messaging.Footer;
import org.apache.qpid.protonj2.types.messaging.Header;
import org.apache.qpid.protonj2.types.messaging.MessageAnnotations;
import org.apache.qpid.protonj2.types.messaging.Properties;
import org.apache.qpid.protonj2.types.messaging.Section;

public class ExternalMessage<E> implements Message<E> {

    private Header header;
    private MessageAnnotations messageAnnotations;
    private Properties properties;
    private ApplicationProperties applicationProperties;
    private Footer footer;
    private Section<E> body;

    private final boolean allowAdvancedConversions;

    public ExternalMessage() {
        this(false);
    }

    public ExternalMessage(boolean allowAdvancedConversions) {
        this(null, allowAdvancedConversions);
    }

    public ExternalMessage(Section<E> section) {
        this(section, false);
    }

    public ExternalMessage(Section<E> body, boolean allowAdvancedConversions) {
        this.body = body;
        this.allowAdvancedConversions = allowAdvancedConversions;
    }

    @Override
    public AdvancedMessage<E> toAdvancedMessage() {
        if (allowAdvancedConversions) {
            return new AdvancedExternalMessage<E>(this);
        } else {
            throw new UnsupportedOperationException("Test ExternalMessage doesn't support AdvancedMessage conversions");
        }
    }

    //----- Entry point for creating new ClientMessage instances.

    public static <V> Message<V> create(Section<V> body) {
        return new ExternalMessage<V>(body);
    }

    //----- Message Header API

    @Override
    public boolean durable() {
        return header == null ? Header.DEFAULT_DURABILITY : header.isDurable();
    }

    @Override
    public ExternalMessage<E> durable(boolean durable) {
        lazyCreateHeader().setDurable(durable);
        return this;
    }

    @Override
    public byte priority() {
        return header == null ? Header.DEFAULT_PRIORITY : header.getPriority();
    }

    @Override
    public ExternalMessage<E> priority(byte priority) {
        lazyCreateHeader().setPriority(priority);
        return this;
    }

    @Override
    public long timeToLive() {
        return header == null ? Header.DEFAULT_TIME_TO_LIVE : header.getTimeToLive();
    }

    @Override
    public ExternalMessage<E> timeToLive(long timeToLive) {
        lazyCreateHeader().setTimeToLive(timeToLive);
        return this;
    }

    @Override
    public boolean firstAcquirer() {
        return header == null ? Header.DEFAULT_FIRST_ACQUIRER : header.isFirstAcquirer();
    }

    @Override
    public ExternalMessage<E> firstAcquirer(boolean firstAcquirer) {
        lazyCreateHeader().setFirstAcquirer(firstAcquirer);
        return this;
    }

    @Override
    public long deliveryCount() {
        return header == null ? Header.DEFAULT_DELIVERY_COUNT : header.getDeliveryCount();
    }

    @Override
    public ExternalMessage<E> deliveryCount(long deliveryCount) {
        lazyCreateHeader().setDeliveryCount(deliveryCount);
        return this;
    }

    //----- Message Properties access

    @Override
    public Object messageId() {
        return properties != null ? properties.getMessageId() : null;
    }

    @Override
    public Message<E> messageId(Object messageId) {
        lazyCreateProperties().setMessageId(messageId);
        return this;
    }

    @Override
    public byte[] userId() {
        if (properties != null && properties.getUserId() != null) {
            return properties.getUserId().asByteArray();
        } else {
            return null;
        }
    }

    @Override
    public Message<E> userId(byte[] userId) {
        lazyCreateProperties().setUserId(new Binary(Arrays.copyOf(userId, userId.length)));
        return this;
    }

    @Override
    public String to() {
        return properties != null ? properties.getTo() : null;
    }

    @Override
    public Message<E> to(String to) {
        lazyCreateProperties().setTo(to);
        return this;
    }

    @Override
    public String subject() {
        return properties != null ? properties.getSubject() : null;
    }

    @Override
    public Message<E> subject(String subject) {
        lazyCreateProperties().setSubject(subject);
        return this;
    }

    @Override
    public String replyTo() {
        return properties != null ? properties.getReplyTo() : null;
    }

    @Override
    public Message<E> replyTo(String replyTo) {
        lazyCreateProperties().setReplyTo(replyTo);
        return this;
    }

    @Override
    public Object correlationId() {
        return properties != null ? properties.getCorrelationId() : null;
    }

    @Override
    public Message<E> correlationId(Object correlationId) {
        lazyCreateProperties().setCorrelationId(correlationId);
        return this;
    }

    @Override
    public String contentType() {
        return properties != null ? properties.getContentType() : null;
    }

    @Override
    public Message<E> contentType(String contentType) {
        lazyCreateProperties().setContentType(contentType);
        return this;
    }

    @Override
    public String contentEncoding() {
        return properties != null ? properties.getContentEncoding() : null;
    }

    @Override
    public Message<E> contentEncoding(String contentEncoding) {
        lazyCreateProperties().setContentEncoding(contentEncoding);
        return this;
    }

    @Override
    public long absoluteExpiryTime() {
        return properties != null ? properties.getAbsoluteExpiryTime() : 0;
    }

    @Override
    public Message<E> absoluteExpiryTime(long expiryTime) {
        lazyCreateProperties().setAbsoluteExpiryTime(expiryTime);
        return this;
    }

    @Override
    public long creationTime() {
        return properties != null ? properties.getCreationTime() : 0;
    }

    @Override
    public Message<E> creationTime(long createTime) {
        lazyCreateProperties().setCreationTime(createTime);
        return this;
    }

    @Override
    public String groupId() {
        return properties != null ? properties.getGroupId() : null;
    }

    @Override
    public Message<E> groupId(String groupId) {
        lazyCreateProperties().setGroupId(groupId);
        return this;
    }

    @Override
    public int groupSequence() {
        return properties != null ? (int) properties.getGroupSequence() : 0;
    }

    @Override
    public Message<E> groupSequence(int groupSequence) {
        lazyCreateProperties().setGroupSequence(groupSequence);
        return this;
    }

    @Override
    public String replyToGroupId() {
        return properties != null ? properties.getReplyToGroupId() : null;
    }

    @Override
    public Message<E> replyToGroupId(String replyToGroupId) {
        lazyCreateProperties().setReplyToGroupId(replyToGroupId);
        return this;
    }

    //----- Message Annotations Access

    @Override
    public Object annotation(String key) {
        Object value = null;
        if (messageAnnotations != null) {
            value = messageAnnotations.getValue().get(Symbol.valueOf(key));
        }

        return value;
    }

    @Override
    public boolean hasAnnotation(String key) {
        if (messageAnnotations != null && messageAnnotations.getValue() != null) {
            return messageAnnotations.getValue().containsKey(Symbol.valueOf(key));
        } else {
            return false;
        }
    }

    @Override
    public boolean hasAnnotations() {
        return messageAnnotations != null &&
               messageAnnotations.getValue() != null &&
               messageAnnotations.getValue().size() > 0;
    }

    @Override
    public Object removeAnnotation(String key) {
        if (hasAnnotations()) {
            return messageAnnotations.getValue().remove(Symbol.valueOf(key));
        } else {
            return null;
        }
     }

    @Override
    public Message<E> forEachAnnotation(BiConsumer<String, Object> action) {
        if (hasAnnotations()) {
            messageAnnotations.getValue().forEach((key, value) -> {
                action.accept(key.toString(), value);
            });
        }

        return this;
    }

    @Override
    public ExternalMessage<E> annotation(String key, Object value) {
        lazyCreateMessageAnnotations().getValue().put(Symbol.valueOf(key),value);
        return this;
    }

    //----- Application Properties Access

    @Override
    public Object property(String key) {
        Object value = null;
        if (applicationProperties != null) {
            value = applicationProperties.getValue().get(key);
        }

        return value;
    }

    @Override
    public boolean hasProperty(String key) {
        if (applicationProperties != null && applicationProperties.getValue() != null) {
            return applicationProperties.getValue().containsKey(key);
        } else {
            return false;
        }
    }

    @Override
    public boolean hasProperties() {
        return applicationProperties != null &&
               applicationProperties.getValue() != null &&
               applicationProperties.getValue().size() > 0;
    }

    @Override
    public Object removeProperty(String key) {
        if (hasProperties()) {
            return applicationProperties.getValue().remove(key);
        } else {
            return null;
        }
     }

    @Override
    public Message<E> forEachProperty(BiConsumer<String, Object> action) {
        if (hasProperties()) {
            applicationProperties.getValue().forEach(action);
        }

        return this;
    }

    @Override
    public ExternalMessage<E> property(String key, Object value) {
        lazyCreateApplicationProperties().getValue().put(key,value);
        return this;
    }

    //----- Footer Access

    @Override
    public Object footer(String key) {
        Object value = null;
        if (footer != null) {
            value = footer.getValue().get(Symbol.valueOf(key));
        }

        return value;
    }

    @Override
    public boolean hasFooter(String key) {
        if (footer != null && footer.getValue() != null) {
            return footer.getValue().containsKey(Symbol.valueOf(key));
        } else {
            return false;
        }
    }

    @Override
    public boolean hasFooters() {
        return footer != null &&
               footer.getValue() != null &&
               footer.getValue().size() > 0;
    }

    @Override
    public Object removeFooter(String key) {
        if (hasFooters()) {
            return footer.getValue().remove(Symbol.valueOf(key));
        } else {
            return null;
        }
     }

    @Override
    public Message<E> forEachFooter(BiConsumer<String, Object> action) {
        if (hasFooters()) {
            footer.getValue().forEach((key, value) -> {
                action.accept(key.toString(), value);
            });
        }

        return this;
    }

    @Override
    public ExternalMessage<E> footer(String key, Object value) {
        lazyCreateFooter().getValue().put(Symbol.valueOf(key),value);
        return this;
    }

    //----- Message body access

    @Override
    public E body() {
        return body.getValue();
    }

    @Override
    public ExternalMessage<E> body(E body) {
        this.body = ClientMessageSupport.createSectionFromValue(body);
        return this;
    }

    public ExternalMessage<E> body(Section<E> body) {
        this.body = body;
        return this;
    }

    //----- Internal API

    private Header lazyCreateHeader() {
        if (header == null) {
            header = new Header();
        }

        return header;
    }

    private Properties lazyCreateProperties() {
        if (properties == null) {
            properties = new Properties();
        }

        return properties;
    }

    private ApplicationProperties lazyCreateApplicationProperties() {
        if (applicationProperties == null) {
            applicationProperties = new ApplicationProperties(new LinkedHashMap<>());
        }

        return applicationProperties;
    }

    private MessageAnnotations lazyCreateMessageAnnotations() {
        if (messageAnnotations == null) {
            messageAnnotations = new MessageAnnotations(new LinkedHashMap<>());
        }

        return messageAnnotations;
    }

    private Footer lazyCreateFooter() {
        if (footer == null) {
            footer = new Footer(new LinkedHashMap<>());
        }

        return footer;
    }

    //----- Sealed AdvancedMessage implementation that wraps this type.

    private static class AdvancedExternalMessage<E> extends ExternalMessage<E> implements AdvancedMessage<E> {

        private final ExternalMessage<E> message;

        private final List<Section<?>> bodySections = new ArrayList<>();
        private int messageFormat;

        /**
         * Create a wrapper that exposes {@link ExternalMessage} as an {@link AdvancedMessage}
         *
         * @param message
         *      this message to wrap.
         */
        public AdvancedExternalMessage(ExternalMessage<E> message) {
            this.message = message;
        }

        @Override
        public Header header() {
            return message.header;
        }

        @Override
        public AdvancedMessage<E> header(Header header) {
            message.header = header;
            return this;
        }

        @Override
        public MessageAnnotations annotations() {
            return message.messageAnnotations;
        }

        @Override
        public AdvancedMessage<E> annotations(MessageAnnotations messageAnnotations) {
            message.messageAnnotations = messageAnnotations;
            return this;
        }

        @Override
        public Properties properties() {
            return message.properties;
        }

        @Override
        public AdvancedMessage<E> properties(Properties properties) {
            message.properties = properties;
            return this;
        }

        @Override
        public ApplicationProperties applicationProperties() {
            return message.applicationProperties;
        }

        @Override
        public AdvancedMessage<E> applicationProperties(ApplicationProperties applicationProperties) {
            message.applicationProperties = applicationProperties;
            return this;
        }

        @Override
        public Footer footer() {
            return message.footer;
        }

        @Override
        public AdvancedMessage<E> footer(Footer footer) {
            message.footer = footer;
            return this;
        }

        @Override
        public int messageFormat() {
            return messageFormat;
        }

        @Override
        public AdvancedMessage<E> messageFormat(int messageFormat) {
            this.messageFormat = messageFormat;
            return this;
        }

        @Override
        public ProtonBuffer encode(Map<String, Object> deliveryAnnotations, ProtonBufferAllocator allocator) throws ClientException {
            return ClientMessageSupport.encodeMessage(this, deliveryAnnotations);
        }

        @Override
        public AdvancedMessage<E> addBodySection(Section<?> bodySection) {
            Objects.requireNonNull(bodySection, "Additional Body Section cannot be null");

            if (bodySections.isEmpty()) {
                // Preserve older section from original message creation.
                if (message.body() != null) {
                    bodySections.add(message.body);
                }

                message.body = null;
            }

            return this;
        }

        @Override
        public AdvancedMessage<E> bodySections(Collection<Section<?>> sections) {
            this.bodySections.clear();
            this.bodySections.addAll(sections);
            return this;
        }

        @Override
        public Collection<Section<?>> bodySections() {
            return Collections.unmodifiableCollection(bodySections);
        }

        @Override
        public AdvancedMessage<E> forEachBodySection(Consumer<Section<?>> consumer) {
            if (!bodySections.isEmpty()) {
                bodySections.forEach(section -> {
                    consumer.accept(section);
                });
            } else {
                if (message.body != null) {
                    consumer.accept(message.body);
                }
            }

            return this;
        }

        @Override
        public AdvancedMessage<E> clearBodySections() {
            bodySections.clear();
            message.body = null;

            return this;
        }
    }
}
