/*
 * 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.messaginghub.amqperative.impl;

import org.apache.qpid.proton4j.amqp.DeliveryTag;
import org.apache.qpid.proton4j.amqp.messaging.Accepted;
import org.apache.qpid.proton4j.engine.OutgoingDelivery;
import org.messaginghub.amqperative.DeliveryState;
import org.messaginghub.amqperative.Sender;
import org.messaginghub.amqperative.Tracker;
import org.messaginghub.amqperative.futures.ClientFuture;

/**
 * Client outgoing delivery tracker object.
 */
public class ClientTracker implements Tracker {

    private final ClientSender sender;
    private final OutgoingDelivery delivery;

    private final ClientFuture<Tracker> acknowledged;

    private volatile boolean remotelySetted;
    private volatile DeliveryState remoteDeliveryState;

    // TODO - Memory consistency issues are observable if auto settle is on as the
    //        settle and update of the remote and local state happens in the client
    //        thread and not on the calling thread reading the state can return the
    //        not yet updated value.

    /**
     * Create an instance of a client outgoing delivery tracker.
     *
     * @param sender
     *      The sender that was used to send the delivery
     * @param delivery
     *      The proton outgoing delivery object that backs this tracker.
     */
    ClientTracker(ClientSender sender, OutgoingDelivery delivery) {
        this.sender = sender;
        this.delivery = delivery;
        this.acknowledged = sender.session().getFutureFactory().createFuture();
    }

    @Override
    public Sender sender() {
        return sender;
    }

    @Override
    public DeliveryState state() {
        return ClientDeliveryState.fromProtonType(delivery.getState());
    }

    @Override
    public DeliveryState remoteState() {
        return remoteDeliveryState;
    }

    @Override
    public boolean remotelySettled() {
        return remotelySetted;
    }

    @Override
    public ClientTracker accept() {
        delivery.disposition(Accepted.getInstance());
        return this;
    }

    @Override
    public ClientTracker disposition(DeliveryState state, boolean settle) {
        org.apache.qpid.proton4j.amqp.transport.DeliveryState protonState = null;
        if (state != null) {
            try {
                protonState = ((ClientDeliveryState) state).getProtonDeliveryState();
            } catch (ClassCastException ccex) {
                throw new IllegalArgumentException("Unknown DeliveryState type given, no disposition applied to Delivery.");
            }
        }

        sender.disposition(delivery, protonState, settle);
        return this;
    }

    @Override
    public ClientTracker settle() {
        sender.disposition(delivery, null, true);
        return this;
    }

    @Override
    public boolean settled() {
        return delivery.isSettled();
    }

    @Override
    public DeliveryTag tag() {
        return delivery.getTag();
    }

    @Override
    public ClientFuture<Tracker> acknowledgeFuture() {
        return acknowledged;
    }

    //----- Internal Event hooks for delivery updates

    void processDeliveryUpdated(OutgoingDelivery delivery) {
        remotelySetted = delivery.isRemotelySettled();
        remoteDeliveryState = ClientDeliveryState.fromProtonType(delivery.getRemoteState());

        if (delivery.isRemotelySettled()) {
            acknowledged.complete(this);
        }
    }
}

@@ -1550,297 +1550,8 @@
 e;%0A%0A
-    // TODO - Memory consistency issues are observable if auto settle is on as the%0A    //        settle and update of the remote and local state happens in the client%0A    //        thread and not on the calling thread reading the state can return the%0A    //        not yet updated value.%0A%0A
     
@@ -2476,24 +2476,22 @@
         
-deliv
+send
 er
-y
 .disposi
@@ -2495,16 +2495,26 @@
 osition(
+delivery, 
 Accepted
@@ -2527,16 +2527,22 @@
 stance()
+, true
 );%0A     

Fix issue with accept not running on the executor thread.