svn commit: r1095927 - in /activemq/activemq-apollo/trunk: apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/ apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/ apollo-tcp/src/main/java/org/apache/activemq/apollo/transp...

[ch...@apache.org]

Author: chirino
Date: Fri Apr 22 12:47:37 2011
New Revision: 1095927

URL: http://svn.apache.org/viewvc?rev=1095927&view=rev
Log:
Add a full() method to the codecs so that it can be used by the transports.

Removed:
    activemq/activemq-apollo/trunk/apollo-transport/src/main/java/org/apache/activemq/apollo/transport/ObjectStreamProtocolCodec.java
    activemq/activemq-apollo/trunk/apollo-transport/src/main/java/org/apache/activemq/apollo/transport/ObjectStreamProtocolCodecFactory.java
Modified:
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/MultiProtocol.scala
    activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompCodec.scala
    activemq/activemq-apollo/trunk/apollo-tcp/src/main/java/org/apache/activemq/apollo/transport/tcp/TcpTransport.java
    activemq/activemq-apollo/trunk/apollo-transport/src/main/java/org/apache/activemq/apollo/transport/ProtocolCodec.java

Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/MultiProtocol.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/MultiProtocol.scala?rev=1095927&r1=1095926&r2=1095927&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/MultiProtocol.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/protocol/MultiProtocol.scala Fri Apr 22 12:47:37 2011
@@ -26,6 +26,7 @@ import java.io.IOException
 import java.lang.String
 import java.util.concurrent.TimeUnit
 import org.fusesource.hawtdispatch._
+import org.apache.activemq.apollo.transport.ProtocolCodec.BufferState
 
 /**
  * <p>
@@ -122,9 +123,11 @@ class MultiProtocolCodec(val protocols: 
 
   def setWritableByteChannel(channel: WritableByteChannel) = {}
 
-  def write(value: Any) = throw new UnsupportedOperationException()
+  def write(value: Any) = ProtocolCodec.BufferState.FULL
 
-  def flush = ProtocolCodec.BufferState.EMPTY
+  def full: Boolean = true
+
+  def flush = ProtocolCodec.BufferState.FULL
 
   def getWriteCounter = 0L
 

Modified: activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompCodec.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompCodec.scala?rev=1095927&r1=1095926&r2=1095927&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompCodec.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompCodec.scala Fri Apr 22 12:47:37 2011
@@ -183,8 +183,8 @@ class StompCodec extends ProtocolCodec {
   var write_direct:ZeroCopyBuffer = null
   var write_direct_pos = 0
 
-  def is_full = next_write_direct!=null || next_write_buffer.size() >= (write_buffer_size >> 2)
-  def is_empty = write_buffer.remaining() == 0 && write_direct==null
+  def full = next_write_direct!=null || next_write_buffer.size >= (write_buffer_size >> 1)
+  def is_empty = write_buffer.remaining == 0 && write_direct==null
 
   def setWritableByteChannel(channel: WritableByteChannel) = {
     this.write_channel = channel
@@ -197,7 +197,7 @@ class StompCodec extends ProtocolCodec {
 
 
   def write(command: Any):ProtocolCodec.BufferState =  {
-    if ( is_full) {
+    if ( full) {
       ProtocolCodec.BufferState.FULL
     } else {
       val was_empty = is_empty

Modified: activemq/activemq-apollo/trunk/apollo-tcp/src/main/java/org/apache/activemq/apollo/transport/tcp/TcpTransport.java
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-tcp/src/main/java/org/apache/activemq/apollo/transport/tcp/TcpTransport.java?rev=1095927&r1=1095926&r2=1095927&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-tcp/src/main/java/org/apache/activemq/apollo/transport/tcp/TcpTransport.java (original)
+++ activemq/activemq-apollo/trunk/apollo-tcp/src/main/java/org/apache/activemq/apollo/transport/tcp/TcpTransport.java Fri Apr 22 12:47:37 2011
@@ -177,7 +177,6 @@ public class TcpTransport extends JavaBa
     private DispatchSource writeSource;
 
     protected boolean useLocalHost = true;
-    protected boolean full = false;
 
     int max_read_rate;
     int max_write_rate;
@@ -478,7 +477,7 @@ public class TcpTransport extends JavaBa
 
 
     public boolean full() {
-        return full;
+        return codec.full();
     }
 
     public boolean offer(Object command) {

Modified: activemq/activemq-apollo/trunk/apollo-transport/src/main/java/org/apache/activemq/apollo/transport/ProtocolCodec.java
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-transport/src/main/java/org/apache/activemq/apollo/transport/ProtocolCodec.java?rev=1095927&r1=1095926&r2=1095927&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-transport/src/main/java/org/apache/activemq/apollo/transport/ProtocolCodec.java (original)
+++ activemq/activemq-apollo/trunk/apollo-transport/src/main/java/org/apache/activemq/apollo/transport/ProtocolCodec.java Fri Apr 22 12:47:37 2011
@@ -100,6 +100,12 @@ public interface ProtocolCodec {
     BufferState flush() throws IOException;
 
     /**
+     * Is the codec's buffer full?
+     * @return
+     */
+    boolean full();
+
+    /**
      * @return The number of bytes written.
      */
     public long getWriteCounter();