activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tab...@apache.org
Subject svn commit: r796314 - in /activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport: TransportFilter.cpp TransportFilter.h inactivity/ inactivity/InactivityMonitor.cpp inactivity/InactivityMonitor.h
Date Tue, 21 Jul 2009 14:21:10 GMT
Author: tabish
Date: Tue Jul 21 14:21:09 2009
New Revision: 796314

URL: http://svn.apache.org/viewvc?rev=796314&view=rev
Log:
Start work on an Inactivity Monitor for the Transports.

Added:
    activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/
    activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.cpp
  (with props)
    activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.h
  (with props)
Modified:
    activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/TransportFilter.cpp
    activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/TransportFilter.h

Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/TransportFilter.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/TransportFilter.cpp?rev=796314&r1=796313&r2=796314&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/TransportFilter.cpp
(original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/TransportFilter.cpp
Tue Jul 21 14:21:09 2009
@@ -32,12 +32,85 @@
 }
 
 ////////////////////////////////////////////////////////////////////////////////
-void TransportFilter::onException( const decaf::lang::Exception& ex ) {
+void TransportFilter::onCommand( const Pointer<Command>& command ){
+    fire( command );
+}
 
+////////////////////////////////////////////////////////////////////////////////
+void TransportFilter::onException( const decaf::lang::Exception& ex ) {
     fire( ex );
 }
 
 ////////////////////////////////////////////////////////////////////////////////
+void TransportFilter::fire( const decaf::lang::Exception& ex ){
+
+    if( listener != NULL ){
+        try{
+            listener->onException( ex );
+        }catch( ... ){}
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void TransportFilter::fire( const Pointer<Command>& command ){
+    try{
+        if( listener != NULL ){
+            listener->onCommand( command );
+        }
+    }catch( ... ){}
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void TransportFilter::transportInterrupted() {
+    try{
+        if( listener != NULL ){
+            listener->transportInterrupted();
+        }
+    }catch( ... ){}
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void TransportFilter::transportResumed() {
+    try{
+        if( listener != NULL ){
+            listener->transportResumed();
+        }
+    }catch( ... ){}
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void TransportFilter::start() throw( cms::CMSException ) {
+
+    if( listener == NULL ){
+        throw exceptions::ActiveMQException( __FILE__, __LINE__,
+            "exceptionListener is invalid" );
+    }
+
+    // Start the delegate transport object.
+    next->start();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void TransportFilter::close() throw( cms::CMSException ) {
+
+    if( next != NULL ) {
+        next->close();
+        next.reset( NULL );
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+Transport* TransportFilter::narrow( const std::type_info& typeId ) {
+    if( typeid( *this ) == typeId ) {
+        return this;
+    } else if( this->next != NULL ) {
+        return this->next->narrow( typeId );
+    }
+
+    return NULL;
+}
+
+////////////////////////////////////////////////////////////////////////////////
 void TransportFilter::reconnect( const decaf::net::URI& uri )
     throw( decaf::io::IOException ) {
 

Modified: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/TransportFilter.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/TransportFilter.h?rev=796314&r1=796313&r2=796314&view=diff
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/TransportFilter.h
(original)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/TransportFilter.h
Tue Jul 21 14:21:09 2009
@@ -34,9 +34,10 @@
     using activemq::commands::Response;
 
     /**
-     * A filter on the transport layer.  Transport
-     * filters implement the Transport interface and
-     * optionally delegate calls to another Transport object.
+     * A filter on the transport layer.  Transport filters implement the Transport
+     * interface and optionally delegate calls to another Transport object.
+     *
+     * @since 1.0
      */
     class AMQCPP_API TransportFilter : public Transport,
                                        public TransportListener {
@@ -55,29 +56,16 @@
     protected:
 
         /**
-         * Notify the exception listener
+         * Notify the listener of the thrown Exception.
          * @param ex - the exception to send to listeners
          */
-        void fire( const decaf::lang::Exception& ex ){
-
-            if( listener != NULL ){
-                try{
-                    listener->onException( ex );
-                }catch( ... ){}
-            }
-        }
+        void fire( const decaf::lang::Exception& ex );
 
         /**
-         * Notify the command listener.
+         * Notify the listener of the new incoming Command.
          * @param command - the command to send to the listener
          */
-        void fire( const Pointer<Command>& command ){
-            try{
-                if( listener != NULL ){
-                    listener->onCommand( command );
-                }
-            }catch( ... ){}
-        }
+        void fire( const Pointer<Command>& command );
 
     public:
 
@@ -93,9 +81,7 @@
          * Event handler for the receipt of a command.
          * @param command - the received command object.
          */
-        virtual void onCommand( const Pointer<Command>& command ){
-            fire( command );
-        }
+        virtual void onCommand( const Pointer<Command>& command );
 
         /**
          * Event handler for an exception from a command transport.
@@ -107,12 +93,12 @@
         /**
          * The transport has suffered an interruption from which it hopes to recover
          */
-        virtual void transportInterrupted() {}
+        virtual void transportInterrupted();
 
         /**
          * The transport has resumed after an interruption
          */
-        virtual void transportResumed() {}
+        virtual void transportResumed();
 
         /**
          * Sends a one-way command.  Does not wait for any response from the
@@ -188,16 +174,7 @@
          * @throws CMSException if an error occurs or if this transport
          * has already been closed.
          */
-        virtual void start() throw( cms::CMSException ) {
-
-            if( listener == NULL ){
-                throw exceptions::ActiveMQException( __FILE__, __LINE__,
-                    "exceptionListener is invalid" );
-            }
-
-            // Start the delegate transport object.
-            next->start();
-        }
+        virtual void start() throw( cms::CMSException );
 
         /**
          * Stops the polling thread and closes the streams.  This can
@@ -205,12 +182,7 @@
          * this object has been closed, it cannot be restarted.
          * @throws CMSException if errors occur.
          */
-        virtual void close() throw( cms::CMSException ){
-            if( next != NULL ) {
-                next->close();
-                next.reset( NULL );
-            }
-        }
+        virtual void close() throw( cms::CMSException );
 
         /**
          * Narrows down a Chain of Transports to a specific Transport to allow a
@@ -221,15 +193,7 @@
          *
          * @return the requested Object. or NULL if its not in this chain.
          */
-        virtual Transport* narrow( const std::type_info& typeId ) {
-            if( typeid( *this ) == typeId ) {
-                return this;
-            } else if( this->next != NULL ) {
-                return this->next->narrow( typeId );
-            }
-
-            return NULL;
-        }
+        virtual Transport* narrow( const std::type_info& typeId );
 
         /**
          * Is this Transport fault tolerant, meaning that it will reconnect to

Added: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.cpp
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.cpp?rev=796314&view=auto
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.cpp
(added)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.cpp
Tue Jul 21 14:21:09 2009
@@ -0,0 +1,58 @@
+/*
+ * 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.
+ */
+
+#include "InactivityMonitor.h"
+
+using namespace std;
+using namespace activemq;
+using namespace activemq::transport;
+using namespace activemq::transport::inactivity;
+using namespace activemq::exceptions;
+using namespace decaf;
+using namespace decaf::io;
+using namespace decaf::util;
+using namespace decaf::util::concurrent;
+using namespace decaf::util::concurrent::atomic;
+using namespace decaf::lang;
+using namespace decaf::lang::exceptions;
+
+////////////////////////////////////////////////////////////////////////////////
+InactivityMonitor::InactivityMonitor( const Pointer<Transport>& next )
+:  TransportFilter( next ) {
+
+}
+
+////////////////////////////////////////////////////////////////////////////////
+InactivityMonitor::~InactivityMonitor() {
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void InactivityMonitor::onException( const decaf::lang::Exception& ex ) {
+    TransportFilter::onException( ex );
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void InactivityMonitor::onCommand( const Pointer<Command>& command ) {
+    TransportFilter::onCommand( command );
+}
+
+////////////////////////////////////////////////////////////////////////////////
+void InactivityMonitor::oneway( const Pointer<Command>& command )
+    throw( decaf::io::IOException, decaf::lang::exceptions::UnsupportedOperationException
) {
+
+    TransportFilter::oneway( command );
+}

Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.cpp
------------------------------------------------------------------------------
    svn:eol-style = native

Added: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.h
URL: http://svn.apache.org/viewvc/activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.h?rev=796314&view=auto
==============================================================================
--- activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.h
(added)
+++ activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.h
Tue Jul 21 14:21:09 2009
@@ -0,0 +1,87 @@
+/*
+ * 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.
+ */
+
+#ifndef _ACTIVEMQ_TRANSPORT_INACTIVITY_INACTIVITYMONITOR_H_
+#define _ACTIVEMQ_TRANSPORT_INACTIVITY_INACTIVITYMONITOR_H_
+
+#include <activemq/util/Config.h>
+
+#include <activemq/transport/TransportFilter.h>
+#include <activemq/commands/Command.h>
+#include <activemq/commands/Response.h>
+#include <activemq/commands/WireFormatInfo.h>
+
+#include <decaf/lang/Pointer.h>
+#include <decaf/util/Timer.h>
+#include <decaf/util/concurrent/atomic/AtomicBoolean.h>
+
+namespace activemq {
+namespace transport {
+namespace inactivity {
+
+    using decaf::lang::Pointer;
+
+    class ReadChecker;
+    class WriteChecker;
+
+    class AMQCPP_API InactivityMonitor : public TransportFilter {
+    private:
+
+        Pointer<commands::WireFormatInfo> localWireFormatInfo;
+        Pointer<commands::WireFormatInfo> remoteWireFormatInfo;
+
+        decaf::util::Timer readCheckTimer;
+        decaf::util::Timer writeCheckTimer;
+
+        decaf::util::concurrent::atomic::AtomicBoolean commandSent;
+        decaf::util::concurrent::atomic::AtomicBoolean commandReceived;
+
+        decaf::util::concurrent::atomic::AtomicBoolean failed;
+        decaf::util::concurrent::atomic::AtomicBoolean inRead;
+        decaf::util::concurrent::atomic::AtomicBoolean inWrite;
+
+        long long readCheckTime;
+        long long writeCheckTime;
+        long long initialDelayTime;
+
+        friend class ReadChecker;
+        friend class WriteChecker;
+
+    public:
+
+        /**
+         * Constructor
+         *
+         * @param next
+         *      The Transport instance that this TransportFilter wraps.
+         */
+        InactivityMonitor( const Pointer<Transport>& next );
+
+        virtual ~InactivityMonitor();
+
+        virtual void onException( const decaf::lang::Exception& ex );
+
+        virtual void onCommand( const Pointer<Command>& command );
+
+        virtual void oneway( const Pointer<Command>& command )
+            throw( decaf::io::IOException, decaf::lang::exceptions::UnsupportedOperationException
);
+
+    };
+
+}}}
+
+#endif /* _ACTIVEMQ_TRANSPORT_INACTIVITY_INACTIVITYMONITOR_H_ */

Propchange: activemq/activemq-cpp/trunk/activemq-cpp/src/main/activemq/transport/inactivity/InactivityMonitor.h
------------------------------------------------------------------------------
    svn:eol-style = native



Mime
View raw message