svn commit: r1209544 - in /activemq/activemq-apollo/trunk: apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/ apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/ apollo-openwire/src/test/scala/org/apache/activemq/a...

[ch...@apache.org]

Author: chirino
Date: Fri Dec  2 15:47:36 2011
New Revision: 1209544

URL: http://svn.apache.org/viewvc?rev=1209544&view=rev
Log:
Fixes APLO-93 : Support initial read/write interval in heartbeat monitor

Patch provided by Stan Lewis.  Thanks!

Modified:
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/HeartBeatMonitor.scala
    activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala
    activemq/activemq-apollo/trunk/apollo-openwire/src/test/scala/org/apache/activemq/apollo/openwire/OpenwireTestSupport.scala

Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/HeartBeatMonitor.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/HeartBeatMonitor.scala?rev=1209544&r1=1209543&r2=1209544&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/HeartBeatMonitor.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/HeartBeatMonitor.scala Fri Dec  2 15:47:36 2011
@@ -29,6 +29,8 @@ import java.util.concurrent.TimeUnit
 class HeartBeatMonitor() {
 
   var transport:Transport = _
+  var initial_write_check_delay = 0L
+  var initial_read_check_delay = 0L
   var write_interval = 0L
   var read_interval = 0L
 
@@ -64,10 +66,22 @@ class HeartBeatMonitor() {
   def start = {
     session += 1
     if( write_interval!=0 ) {
-      schedual_check_writes(session)
+      if ( initial_write_check_delay!=0 ) {
+        transport.getDispatchQueue.after(initial_write_check_delay, TimeUnit.MILLISECONDS) {
+          schedual_check_writes(session)
+        }
+      } else {
+        schedual_check_writes(session)
+      }
     }
     if( read_interval!=0 ) {
-      schedual_check_reads(session)
+      if ( initial_read_check_delay!=0 ) {
+        transport.getDispatchQueue.after(initial_read_check_delay, TimeUnit.MILLISECONDS) {
+          schedual_check_reads(session)
+        }
+      } else {
+        schedual_check_reads(session)
+      }
     }
   }
 

Modified: activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala?rev=1209544&r1=1209543&r2=1209544&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala Fri Dec  2 15:47:36 2011
@@ -387,10 +387,6 @@ class OpenwireProtocolHandler extends Pr
     val inactive_time = preferred_wireformat_settings.getMaxInactivityDuration().min(info.getMaxInactivityDuration())
     val initial_delay = preferred_wireformat_settings.getMaxInactivityDurationInitalDelay().min(info.getMaxInactivityDurationInitalDelay())
 
-    if (initial_delay != inactive_time) {
-      die("We only support an initial delay inactivity duration equal to the max inactivity duration")
-    }
-
     if (inactive_time > 0) {
       heart_beat_monitor.read_interval = inactive_time
       // lets be a little forgiving to account to packet transmission latency.
@@ -408,6 +404,9 @@ class OpenwireProtocolHandler extends Pr
       }
     }
 
+    heart_beat_monitor.initial_read_check_delay = initial_delay
+    heart_beat_monitor.initial_write_check_delay = initial_delay
+
     heart_beat_monitor.transport = connection.transport
     heart_beat_monitor.start
 

Modified: activemq/activemq-apollo/trunk/apollo-openwire/src/test/scala/org/apache/activemq/apollo/openwire/OpenwireTestSupport.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-openwire/src/test/scala/org/apache/activemq/apollo/openwire/OpenwireTestSupport.scala?rev=1209544&r1=1209543&r2=1209544&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-openwire/src/test/scala/org/apache/activemq/apollo/openwire/OpenwireTestSupport.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-openwire/src/test/scala/org/apache/activemq/apollo/openwire/OpenwireTestSupport.scala Fri Dec  2 15:47:36 2011
@@ -35,7 +35,7 @@ class OpenwireTestSupport extends FunSui
   val broker_config_uri = "xml:classpath:apollo-openwire.xml"
   val transport_scheme = "tcp"
   val transport_host = "localhost"
-  val uri_options = "?wireFormat.maxInactivityDuration=1000000&wireFormat.maxInactivityDurationInitalDelay=1000000"
+  val uri_options = ""//"?wireFormat.maxInactivityDuration=1000000&wireFormat.maxInactivityDurationInitalDelay=1000000"
 
   override protected def beforeAll() {
     info("Loading broker configuration from the classpath with URI: " + broker_config_uri)