/**
 *
 * Copyright 2005-2006 The Apache Software Foundation
 *
 * 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.
 */
package org.apache.activemq.transport;

import java.io.IOException;

import org.apache.activemq.command.Command;
import org.apache.activemq.command.KeepAliveInfo;
import org.apache.activemq.thread.Scheduler;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;

/**
 * Used to make sure that commands are arriving periodically from the peer of the transport.  
 * 
 * @version $Revision$
 */
public class InactivityMonitor extends TransportFilter implements Runnable {

    private final Log log = LogFactory.getLog(InactivityMonitor.class);
    
    private final long maxInactivityDuration;
    private final AtomicBoolean cancled = new AtomicBoolean(false);
    private byte readCheckIteration=0;

    private final AtomicBoolean commandSent=new AtomicBoolean(true);
    private final AtomicBoolean inSend=new AtomicBoolean(false);

    private final AtomicBoolean commandReceived=new AtomicBoolean(true);
    private final AtomicBoolean inReceive=new AtomicBoolean(false);
    
    public InactivityMonitor(Transport next, long maxInactivityDuration) {
        super(next);
        this.maxInactivityDuration = maxInactivityDuration;
    }
        
    public void start() throws Exception {
        next.start();
        Scheduler.executePeriodically(this, maxInactivityDuration/2);
    }
    
    public void stop() throws Exception {
        if( cancled.compareAndSet(false, true) ) {
            Scheduler.cancel(this);
        }
        next.stop();
    }
    
    public void run() {
        switch(readCheckIteration) {
        case 0:
            writeCheck();
            readCheckIteration++;
        case 1:
            readCheck();
            writeCheck();
            readCheckIteration=0;
            break;
        }        
    }
    
    private void writeCheck() {
        if( inSend.get() ) {
            log.debug("A send is in progress");
            return;
        }
        
        if( !commandSent.get() ) {
            log.debug("No message sent since last write check, sending a KeepAliveInfo");
            try {
                next.oneway(new KeepAliveInfo());
            } catch (IOException e) {
                onException(e);
            }
        } else {
            log.debug("Message sent since last write check, resetting flag");
        }
        
        commandSent.set(false);
        
    }

    private void readCheck() {
        if( inReceive.get() ) {
            log.debug("A receive is in progress");
            return;
        }
        
        if( !commandReceived.get() ) {
            log.debug("No message received since last read check!");
            onException(new InactivityIOException("Channel was inactive for too long."));           
        } else {
            log.debug("Message received since last read check, resetting flag");
        }
        
        commandReceived.set(false);
    }

    public void onCommand(Command command) {
        inReceive.set(true);
        try {
            commandListener.onCommand(command);
        } finally {
            inReceive.set(false);
            commandReceived.set(true);
        }
    }
    
    public void oneway(Command command) throws IOException {
        // Disable inactivity monitoring while processing a command.
        inSend.set(true);
        commandSent.set(true);
        try {
            next.oneway(command);
        } finally {
            inSend.set(false);
        }
    }
    
    public void onException(IOException error) {
        if( cancled.compareAndSet(false, true) ) {
            Scheduler.cancel(this);
        }
        commandListener.onException(error);
    }
}

@@ -1318,76 +1318,8 @@
 on;%0A
-    private final AtomicBoolean cancled = new AtomicBoolean(false);%0A
     
@@ -1408,35 +1408,36 @@
 w AtomicBoolean(
-tru
+fals
 e);%0A    private 
@@ -1999,63 +1999,8 @@
 n %7B%0A
-        if( cancled.compareAndSet(false, true) ) %7B%0A    
     
@@ -2023,34 +2023,24 @@
 ncel(this);%0A
-        %7D%0A
         next
@@ -2055,32 +2055,45 @@
 %0A    %7D%0A    %0A    
+synchronized 
 public void run(
@@ -4077,63 +4077,8 @@
 ) %7B%0A
-        if( cancled.compareAndSet(false, true) ) %7B%0A    
     
@@ -4105,26 +4105,16 @@
 (this);%0A
-        %7D%0A
         

simplified a little.