/*
 * #%L
 * mosaic-connectors
 * %%
 * Copyright (C) 2010 - 2012 Institute e-Austria Timisoara (Romania)
 * %%
 * 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.
 * #L%
 */

package eu.mosaic_cloud.connectors.queue.amqp;

import java.util.UUID;

import eu.mosaic_cloud.connectors.core.ConfigProperties;
import eu.mosaic_cloud.connectors.tools.ConnectorEnvironment;
import eu.mosaic_cloud.platform.core.configuration.ConfigUtils;
import eu.mosaic_cloud.platform.core.configuration.ConfigurationIdentifier;
import eu.mosaic_cloud.platform.core.configuration.IConfiguration;
import eu.mosaic_cloud.platform.core.utils.DataEncoder;
import eu.mosaic_cloud.platform.core.utils.EncodingException;
import eu.mosaic_cloud.platform.interop.common.amqp.AmqpExchangeType;
import eu.mosaic_cloud.platform.interop.common.amqp.AmqpOutboundMessage;
import eu.mosaic_cloud.tools.callbacks.core.CallbackCompletion;
import eu.mosaic_cloud.tools.exceptions.core.FallbackExceptionTracer;

public final class AmqpQueuePublisherConnectorProxy<TMessage> extends
        AmqpQueueConnectorProxy<TMessage> implements
        IAmqpQueuePublisherConnector<TMessage> {

    private final boolean definePassive;
    private final String exchange;
    private final boolean exchangeAutoDelete;
    private final boolean exchangeDurable;
    private final AmqpExchangeType exchangeType;
    private final String identity;
    private final String publishRoutingKey;

    public static <Message> AmqpQueuePublisherConnectorProxy<Message> create(
            final IConfiguration configuration,
            final ConnectorEnvironment environment,
            final Class<Message> messageClass,
            final DataEncoder<Message> messageEncoder) {
        final AmqpQueueRawConnectorProxy rawProxy = AmqpQueueRawConnectorProxy
                .create(configuration, environment);
        final IConfiguration subConfiguration = configuration
                .spliceConfiguration(ConfigurationIdentifier
                        .resolveRelative("publisher"));
        final AmqpQueuePublisherConnectorProxy<Message> proxy = new AmqpQueuePublisherConnectorProxy<Message>(
                rawProxy, subConfiguration, messageClass, messageEncoder);
        return proxy;
    }

    private AmqpQueuePublisherConnectorProxy(
            final AmqpQueueRawConnectorProxy rawProxy,
            final IConfiguration configuration,
            final Class<TMessage> messageClass,
            final DataEncoder<TMessage> messageEncoder) {
        super(rawProxy, configuration, messageClass, messageEncoder);
        this.identity = UUID.randomUUID().toString();
        this.exchange = ConfigUtils
                .resolveParameter(
                        configuration,
                        ConfigProperties.getString("AmqpQueueConnector.0"), String.class, this.identity); //$NON-NLS-1$ 
        this.exchangeType = ConfigUtils
                .resolveParameter(
                        configuration,
                        ConfigProperties.getString("AmqpQueueConnector.5"), AmqpExchangeType.class, AmqpExchangeType.DIRECT);//$NON-NLS-1$
        this.exchangeDurable = ConfigUtils
                .resolveParameter(
                        configuration,
                        ConfigProperties.getString("AmqpQueueConnector.9"), Boolean.class, Boolean.FALSE).booleanValue(); //$NON-NLS-1$ 
        this.exchangeAutoDelete = ConfigUtils
                .resolveParameter(
                        configuration,
                        ConfigProperties.getString("AmqpQueueConnector.7"), Boolean.class, Boolean.TRUE).booleanValue(); //$NON-NLS-1$
        this.publishRoutingKey = ConfigUtils
                .resolveParameter(
                        configuration,
                        ConfigProperties.getString("AmqpQueueConnector.1"), String.class, this.identity); //$NON-NLS-1$ 
        this.definePassive = ConfigUtils
                .resolveParameter(
                        configuration,
                        ConfigProperties.getString("AmqpQueueConnector.8"), Boolean.class, Boolean.FALSE).booleanValue(); //$NON-NLS-1$ 
    }

    @Override
    public CallbackCompletion<Void> destroy() {
        return this.raw.destroy();
    }

    @Override
    public CallbackCompletion<Void> initialize() {
        this.raw.initialize();
        return this.raw.declareExchange(this.exchange, this.exchangeType,
                this.exchangeDurable, this.exchangeAutoDelete,
                this.definePassive);
    }

    @Override
    public CallbackCompletion<Void> publish(final TMessage message) {
        final byte[] data;
        CallbackCompletion<Void> result;
        try {
            data = this.messageEncoder.encode(message);
            final AmqpOutboundMessage outbound = new AmqpOutboundMessage(
                    this.exchange, this.publishRoutingKey, data, false, false,
                    false, null);
            result = this.raw.publish(outbound);
        } catch (final EncodingException exception) {
            FallbackExceptionTracer.defaultInstance
                    .traceDeferredException(exception);
            result = CallbackCompletion.createFailure(exception);
        }

        return result;
    }
}

@@ -5088,22 +5088,16 @@
         
-final 
 byte%5B%5D d
@@ -5099,16 +5099,23 @@
 e%5B%5D data
+ = null
 ;%0A      
@@ -5143,24 +5143,31 @@
 Void%3E result
+ = null
 ;%0A        tr
@@ -5230,244 +5230,8 @@
 e);%0A
-            final AmqpOutboundMessage outbound = new AmqpOutboundMessage(%0A                    this.exchange, this.publishRoutingKey, data, false, false,%0A                    false, null);%0A            result = this.raw.publish(outbound);%0A
     
@@ -5460,24 +5460,287 @@
 ;%0A        %7D%0A
+        if (result == null) %7B%0A%09        final AmqpOutboundMessage outbound = new AmqpOutboundMessage(%0A%09                this.exchange, this.publishRoutingKey, data, false, false,%0A%09                false, null);%0A%09        result = this.raw.publish(outbound);%0A        %7D
 %0A        ret

Fix `AmqpQueuePublisherConnectorProxy` `publish` flow to not delegate the publish inside the decoding `try / catch`