/**
 * 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.cxf.interceptor;

import java.io.IOException;
import java.util.concurrent.Executor;

import org.apache.cxf.Bus;
import org.apache.cxf.endpoint.Endpoint;
import org.apache.cxf.message.Exchange;
import org.apache.cxf.message.Message;
import org.apache.cxf.message.MessageUtils;
import org.apache.cxf.phase.AbstractPhaseInterceptor;
import org.apache.cxf.phase.Phase;
import org.apache.cxf.transport.Conduit;
import org.apache.cxf.workqueue.WorkQueueManager;


/**
 * 
 */
public class OneWayProcessorInterceptor extends AbstractPhaseInterceptor<Message> {
    public OneWayProcessorInterceptor() {
        super(Phase.PRE_LOGICAL);
    }
    public OneWayProcessorInterceptor(String phase) {
        super(phase);
    }
    
    public void handleMessage(Message message) throws Fault {
        
        if (message.getExchange().isOneWay() 
            && !MessageUtils.isRequestor(message)
            && message.get(OneWayProcessorInterceptor.class) == null
            && message.getExchange().get(Executor.class) == null
            && !isSkipRequestAck(message)) { 
            //one way on server side with pre-request ack required, 
            // so fork the rest of this chain onto the
            // workqueue, call the Outgoing chain directly.
            
            message.put(OneWayProcessorInterceptor.class, this);
            final InterceptorChain chain = message.getInterceptorChain();
            try {
                Message partial = createMessage(message.getExchange());
                partial.setExchange(message.getExchange());
                Conduit conduit = message.getExchange().getDestination()
                        .getBackChannel(message, null, null);
                conduit.prepare(partial);
                conduit.close(partial);
            } catch (IOException e) {
                // IGNORE
            }
            
            chain.pause();
            message.getExchange().get(Bus.class).getExtension(WorkQueueManager.class)
                .getAutomaticWorkQueue().execute(new Runnable() {
                    public void run() {
                        chain.resume();
                    }
                });
        }
    }
    
    private static Message createMessage(Exchange exchange) {
        Endpoint ep = exchange.get(Endpoint.class);
        Message msg = null;
        if (ep != null) {
            msg = ep.getBinding().createMessage();
        }
        return msg;
    }

    /**
     * @param message the current message
     * @returned true if server does not need to 
     * send a acknowledgment response to the client's one-way request
     */
    protected static boolean isSkipRequestAck(Message message) {
        Boolean ret = (Boolean)message.get(Message.SKIP_PRE_REQUEST_ACK);
        return ret != null && ret.booleanValue();
    }
}
