/*

 * 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.catalina.tribes.group;

import org.apache.catalina.tribes.ChannelException;

import org.apache.catalina.tribes.ChannelMessage;

import org.apache.catalina.tribes.ChannelReceiver;

import org.apache.catalina.tribes.ChannelSender;

import org.apache.catalina.tribes.Member;

import org.apache.catalina.tribes.MembershipService;

import org.apache.catalina.tribes.MessageListener;

import org.apache.catalina.tribes.transport.SenderState;

import org.apache.catalina.tribes.transport.ReplicationTransmitter;

import org.apache.catalina.tribes.membership.McastService;

import org.apache.catalina.tribes.transport.nio.NioReceiver;

import org.apache.catalina.tribes.Channel;

import org.apache.catalina.tribes.util.Logs;

import org.apache.catalina.tribes.UniqueId;

import org.apache.catalina.tribes.util.Arrays;

/**

 * The channel coordinator object coordinates the membership service,

 * the sender and the receiver.

 * This is the last interceptor in the chain.

 * @author Filip Hanik

 */

public class ChannelCoordinator extends ChannelInterceptorBase implements MessageListener {

    private ChannelReceiver clusterReceiver = new NioReceiver();

    private ChannelSender clusterSender = new ReplicationTransmitter();

    private MembershipService membershipService = new McastService();

    //override optionflag

    protected int optionFlag = Channel.SEND_OPTIONS_BYTE_MESSAGE|Channel.SEND_OPTIONS_USE_ACK|Channel.SEND_OPTIONS_SYNCHRONIZED_ACK;

    public int getOptionFlag() {return optionFlag;}

    public void setOptionFlag(int flag) {optionFlag=flag;}

    private int startLevel = 0;

    public ChannelCoordinator() {

    }

    public ChannelCoordinator(ChannelReceiver receiver,

                              ChannelSender sender,

                              MembershipService service) {

        this();

        this.setClusterReceiver(receiver);

        this.setClusterSender(sender);

        this.setMembershipService(service);

    }

    /**

     * Send a message to one or more members in the cluster

     * @param destination Member[] - the destinations, null or zero length means all

     * @param msg ClusterMessage - the message to send

     * @param options int - sender options, see class documentation

     * @return ClusterMessage[] - the replies from the members, if any.

     */

    public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {

        if ( destination == null ) destination = membershipService.getMembers();

        clusterSender.sendMessage(msg,destination);

        if ( Logs.MESSAGES.isTraceEnabled() ) {

            Logs.MESSAGES.trace("ChannelCoordinator - Sent msg:" + new UniqueId(msg.getUniqueId()) + " at " +new java.sql.Timestamp(System.currentTimeMillis())+ " to "+Arrays.toNameString(destination));

        }

    }

    /**

     * Starts up the channel. This can be called multiple times for individual services to start

     * The svc parameter can be the logical or value of any constants

     * @param svc int value of <BR>

     * DEFAULT - will start all services <BR>

     * MBR_RX_SEQ - starts the membership receiver <BR>

     * MBR_TX_SEQ - starts the membership broadcaster <BR>

     * SND_TX_SEQ - starts the replication transmitter<BR>

     * SND_RX_SEQ - starts the replication receiver<BR>

     * @throws ChannelException if a startup error occurs or the service is already started.

     */

    public void start(int svc) throws ChannelException {

        this.internalStart(svc);

    }

    /**

     * Shuts down the channel. This can be called multiple times for individual services to shutdown

     * The svc parameter can be the logical or value of any constants

     * @param svc int value of <BR>

     * DEFAULT - will shutdown all services <BR>

     * MBR_RX_SEQ - stops the membership receiver <BR>

     * MBR_TX_SEQ - stops the membership broadcaster <BR>

     * SND_TX_SEQ - stops the replication transmitter<BR>

     * SND_RX_SEQ - stops the replication receiver<BR>

     * @throws ChannelException if a startup error occurs or the service is already started.

     */

    public void stop(int svc) throws ChannelException {

        this.internalStop(svc);

    }    

    /**

     * Starts up the channel. This can be called multiple times for individual services to start

     * The svc parameter can be the logical or value of any constants

     * @param svc int value of <BR>

     * DEFAULT - will start all services <BR>

     * MBR_RX_SEQ - starts the membership receiver <BR>

     * MBR_TX_SEQ - starts the membership broadcaster <BR>

     * SND_TX_SEQ - starts the replication transmitter<BR>

     * SND_RX_SEQ - starts the replication receiver<BR>

     * @throws ChannelException if a startup error occurs or the service is already started.

     */

    protected synchronized void internalStart(int svc) throws ChannelException {

        try {

            boolean valid = false;

            //make sure we don't pass down any flags that are unrelated to the bottom layer

            svc = svc & Channel.DEFAULT;

            if (startLevel == Channel.DEFAULT) return; //we have already started up all components

            if (svc == 0 ) return;//nothing to start

            if (svc == (svc & startLevel)) throw new ChannelException("Channel already started for level:"+svc);

            //must start the receiver first so that we can coordinate the port it

            //listens to with the local membership settings

            if ( Channel.SND_RX_SEQ==(svc & Channel.SND_RX_SEQ) ) {

                clusterReceiver.setMessageListener(this);

                clusterReceiver.start();

                //synchronize, big time FIXME

                membershipService.setLocalMemberProperties(getClusterReceiver().getHost(), getClusterReceiver().getPort());

                valid = true;

            }

            if ( Channel.SND_TX_SEQ==(svc & Channel.SND_TX_SEQ) ) {

                clusterSender.start();

                valid = true;

            }

            if ( Channel.MBR_RX_SEQ==(svc & Channel.MBR_RX_SEQ) ) {

                membershipService.setMembershipListener(this);

                membershipService.start(MembershipService.MBR_RX);

                valid = true;

            }

            if ( Channel.MBR_TX_SEQ==(svc & Channel.MBR_TX_SEQ) ) {

                membershipService.start(MembershipService.MBR_TX);

                valid = true;

            }

            if ( !valid) {

                throw new IllegalArgumentException("Invalid start level, valid levels are:SND_RX_SEQ,SND_TX_SEQ,MBR_TX_SEQ,MBR_RX_SEQ");

            }

            startLevel = (startLevel | svc);

        }catch ( ChannelException cx ) {

            throw cx;

        }catch ( Exception x ) {

            throw new ChannelException(x);

        }

    }

    /**

     * Shuts down the channel. This can be called multiple times for individual services to shutdown

     * The svc parameter can be the logical or value of any constants

     * @param svc int value of <BR>

     * DEFAULT - will shutdown all services <BR>

     * MBR_RX_SEQ - starts the membership receiver <BR>

     * MBR_TX_SEQ - starts the membership broadcaster <BR>

     * SND_TX_SEQ - starts the replication transmitter<BR>

     * SND_RX_SEQ - starts the replication receiver<BR>

     * @throws ChannelException if a startup error occurs or the service is already started.

     */

    protected synchronized void internalStop(int svc) throws ChannelException {

        try {

            //make sure we don't pass down any flags that are unrelated to the bottom layer

            svc = svc & Channel.DEFAULT;

            if (startLevel == 0) return; //we have already stopped up all components

            if (svc == 0 ) return;//nothing to stop

            boolean valid = false;

            if ( Channel.SND_RX_SEQ==(svc & Channel.SND_RX_SEQ) ) {

                clusterReceiver.stop();

                clusterReceiver.setMessageListener(null);

                valid = true;

            }

            if ( Channel.SND_TX_SEQ==(svc & Channel.SND_TX_SEQ) ) {

                clusterSender.stop();

                valid = true;

            }

            if ( Channel.MBR_RX_SEQ==(svc & Channel.MBR_RX_SEQ) ) {

                membershipService.stop(MembershipService.MBR_RX);

                membershipService.setMembershipListener(null);

                valid = true;

            }

            if ( Channel.MBR_TX_SEQ==(svc & Channel.MBR_TX_SEQ) ) {

                valid = true;

                membershipService.stop(MembershipService.MBR_TX);

            }            

            if ( !valid) {

                throw new IllegalArgumentException("Invalid start level, valid levels are:SND_RX_SEQ,SND_TX_SEQ,MBR_TX_SEQ,MBR_RX_SEQ");

            }

            startLevel = (startLevel & (~svc));

        }catch ( Exception x ) {

            throw new ChannelException(x);

        } finally {

        }

    }

    public void memberAdded(Member member){

        SenderState.getSenderState(member);

        super.memberAdded(member);

    }

    public void memberDisappeared(Member member){

        SenderState.removeSenderState(member);

        super.memberDisappeared(member);

    }

    public void messageReceived(ChannelMessage msg) {

        if ( Logs.MESSAGES.isTraceEnabled() ) {

            Logs.MESSAGES.trace("ChannelCoordinator - Received msg:" + new UniqueId(msg.getUniqueId()) + " at " +new java.sql.Timestamp(System.currentTimeMillis())+ " from "+msg.getAddress().getName());

        }

        super.messageReceived(msg);

    }

    public ChannelReceiver getClusterReceiver() {

        return clusterReceiver;

    }

    public ChannelSender getClusterSender() {

        return clusterSender;

    }

    public MembershipService getMembershipService() {

        return membershipService;

    }

    public void setClusterReceiver(ChannelReceiver clusterReceiver) {

        if ( clusterReceiver != null ) {

            this.clusterReceiver = clusterReceiver;

            this.clusterReceiver.setMessageListener(this);

        } else {

            if  (this.clusterReceiver!=null ) this.clusterReceiver.setMessageListener(null);

            this.clusterReceiver = null;

        }

    }

    public void setClusterSender(ChannelSender clusterSender) {

        this.clusterSender = clusterSender;

    }

    public void setMembershipService(MembershipService membershipService) {

        this.membershipService = membershipService;

        this.membershipService.setMembershipListener(this);

    }

    public void heartbeat() {

        if ( clusterSender!=null ) clusterSender.heartbeat();

        super.heartbeat();

    }

    /**

     * has members

     */

    public boolean hasMembers() {

        return this.getMembershipService().hasMembers();

    }

    /**

     * Get all current cluster members

     * @return all members or empty array

     */

    public Member[] getMembers() {

        return this.getMembershipService().getMembers();

    }

    /**

     * 

     * @param mbr Member

     * @return Member

     */

    public Member getMember(Member mbr){

        return this.getMembershipService().getMember(mbr);

    }

    /**

     * Return the member that represents this node.

     *

     * @return Member

     */

    public Member getLocalMember(boolean incAlive) {

        return this.getMembershipService().getLocalMember(incAlive);

    }

}