tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fha...@apache.org
Subject svn commit: r628881 - in /tomcat/trunk: java/org/apache/catalina/tribes/ java/org/apache/catalina/tribes/io/ java/org/apache/catalina/tribes/transport/ java/org/apache/catalina/tribes/transport/nio/ test/org/apache/catalina/tribes/test/channel/
Date Mon, 18 Feb 2008 22:07:12 GMT
Author: fhanik
Date: Mon Feb 18 14:07:09 2008
New Revision: 628881

URL: http://svn.apache.org/viewvc?rev=628881&view=rev
Log:
Starting to add in UDP support, still need to rethink how the sender is going to work

Modified:
    tomcat/trunk/java/org/apache/catalina/tribes/Channel.java
    tomcat/trunk/java/org/apache/catalina/tribes/io/ObjectReader.java
    tomcat/trunk/java/org/apache/catalina/tribes/transport/AbstractSender.java
    tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
    tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
    tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioSender.java
    tomcat/trunk/test/org/apache/catalina/tribes/test/channel/ChannelStartStop.java

Modified: tomcat/trunk/java/org/apache/catalina/tribes/Channel.java
URL: http://svn.apache.org/viewvc/tomcat/trunk/java/org/apache/catalina/tribes/Channel.java?rev=628881&r1=628880&r2=628881&view=diff
==============================================================================
--- tomcat/trunk/java/org/apache/catalina/tribes/Channel.java (original)
+++ tomcat/trunk/java/org/apache/catalina/tribes/Channel.java Mon Feb 18 14:07:09 2008
@@ -5,9 +5,9 @@
  * 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.
@@ -56,20 +56,20 @@
  *                                  |
  *                             Coordinator (implements MessageListener,MembershipListener,ChannelInterceptor)
  *                          --------------------
- *                         /        |           \ 
+ *                         /        |           \
  *                        /         |            \
  *                       /          |             \
  *                      /           |              \
  *                     /            |               \
  *           MembershipService ChannelSender ChannelReceiver                        [IO layer]
  * </code></pre>
- * 
+ *
  * For example usage @see org.apache.catalina.tribes.group.GroupChannel
  * @author Filip Hanik
  * @version $Revision$, $Date$
  */
 public interface Channel {
-    
+
     /**
      * Start and stop sequences can be controlled by these constants
      * This allows you to start separate components of the channel <br>
@@ -119,7 +119,7 @@
      * @see #stop(int)
      */
     public static final int MBR_TX_SEQ = 8;
-    
+
     /**
      * Send options, when a message is sent, it can have an option flag
      * to trigger certain behavior. Most flags are used to trigger channel interceptors
@@ -127,7 +127,7 @@
      * However, there are five default flags that every channel implementation must implement<br>
      * SEND_OPTIONS_BYTE_MESSAGE - The message is a pure byte message and no marshalling
or unmarshalling will
      * be performed.<br>
-     * 
+     *
      * @see #send(Member[], Serializable , int)
      * @see #send(Member[], Serializable, int, ErrorHandler)
      */
@@ -150,27 +150,27 @@
      * to trigger certain behavior. Most flags are used to trigger channel interceptors
      * as the message passes through the channel stack. <br>
      * However, there are five default flags that every channel implementation must implement<br>
-     * SEND_OPTIONS_SYNCHRONIZED_ACK - Message is sent and an ACK is received when the message
has been received and 
+     * SEND_OPTIONS_SYNCHRONIZED_ACK - Message is sent and an ACK is received when the message
has been received and
      * processed by the recipient<br>
      * If no ack is received, the message is not considered successful<br>
      * @see #send(Member[], Serializable , int)
      * @see #send(Member[], Serializable, int, ErrorHandler)
      */
     public static final int SEND_OPTIONS_SYNCHRONIZED_ACK = 0x0004;
-    
+
     /**
      * Send options, when a message is sent, it can have an option flag
      * to trigger certain behavior. Most flags are used to trigger channel interceptors
      * as the message passes through the channel stack. <br>
      * However, there are five default flags that every channel implementation must implement<br>
-     * SEND_OPTIONS_ASYNCHRONOUS - Message is sent and an ACK is received when the message
has been received and 
+     * SEND_OPTIONS_ASYNCHRONOUS - Message is sent and an ACK is received when the message
has been received and
      * processed by the recipient<br>
      * If no ack is received, the message is not considered successful<br>
      * @see #send(Member[], Serializable , int)
      * @see #send(Member[], Serializable, int, ErrorHandler)
      */
     public static final int SEND_OPTIONS_ASYNCHRONOUS = 0x0008;
-    
+
     /**
      * Send options, when a message is sent, it can have an option flag
      * to trigger certain behavior. Most flags are used to trigger channel interceptors
@@ -181,7 +181,14 @@
      * @see #send(Member[], Serializable, int, ErrorHandler)
      */
     public static final int SEND_OPTIONS_SECURE = 0x0010;
-    
+
+    /**
+     * Send options. When a message is sent with this flag on
+     * the system sends the message using UDP instead of TCP
+     * @see #send(Member[], Serializable , int)
+     * @see #send(Member[], Serializable, int, ErrorHandler)
+     */
+    public static final int SEND_OPTIONS_UDP =  0x0020;
 
     /**
      * Send options, when a message is sent, it can have an option flag
@@ -196,13 +203,13 @@
      */
     public static final int SEND_OPTIONS_DEFAULT = SEND_OPTIONS_USE_ACK;
 
-    
+
     /**
      * Adds an interceptor to the channel message chain.
      * @param interceptor ChannelInterceptor
      */
     public void addInterceptor(ChannelInterceptor interceptor);
-    
+
     /**
      * 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
@@ -212,7 +219,7 @@
      * MBR_TX_SEQ - starts the membership broadcaster <BR>
      * SND_TX_SEQ - starts the replication transmitter<BR>
      * SND_RX_SEQ - starts the replication receiver<BR>
-     * <b>Note:</b> In order for the membership broadcaster to 
+     * <b>Note:</b> In order for the membership broadcaster to
      * transmit the correct information, it has to be started after the replication receiver.
      * @throws ChannelException if a startup error occurs or the service is already started
or an error occurs.
      */
@@ -229,14 +236,14 @@
      * SND_RX_SEQ - stops the replication receiver<BR>
      * @throws ChannelException if a startup error occurs or the service is already stopped
or an error occurs.
      */
-    public void stop(int svc) throws ChannelException;    
-    
+    public void stop(int svc) throws ChannelException;
+
     /**
      * Send a message to one or more members in the cluster
      * @param destination Member[] - the destinations, can not be null or zero length, the
reason for that
      * is that a membership change can occur and at that time the application is uncertain
what group the message
      * actually got sent to.
-     * @param msg Serializable - the message to send, has to be serializable, or a <code>ByteMessage</code>
to 
+     * @param msg Serializable - the message to send, has to be serializable, or a <code>ByteMessage</code>
to
      * send a pure byte array
      * @param options int - sender options, see class documentation for each interceptor
that is configured in order to trigger interceptors
      * @return a unique Id that identifies the message that is sent
@@ -257,10 +264,10 @@
      * @exception ChannelException - if a serialization error happens.
      */
     public UniqueId send(Member[] destination, Serializable msg, int options, ErrorHandler
handler) throws ChannelException;
-    
+
     /**
      * Sends a heart beat through the interceptor stacks
-     * Use this method to alert interceptors and other components to 
+     * Use this method to alert interceptors and other components to
      * clean up garbage, timed out messages etc.<br>
      * If you application has a background thread, then you can save one thread,
      * by configuring your channel to not use an internal heartbeat thread
@@ -268,14 +275,14 @@
      * @see #setHeartbeat(boolean)
      */
     public void heartbeat();
-    
+
     /**
      * Enables or disables internal heartbeat.
      * @param enable boolean - default value is implementation specific
      * @see #heartbeat()
      */
     public void setHeartbeat(boolean enable);
-    
+
     /**
      * Add a membership listener, will get notified when a new member joins, leaves or crashes
      * <br>If the membership listener implements the Heartbeat interface
@@ -284,7 +291,7 @@
      * @see MembershipListener
      */
     public void addMembershipListener(MembershipListener listener);
-    
+
     /**
      * Add a channel listener, this is a callback object when messages are received
      * <br>If the channel listener implements the Heartbeat interface
@@ -307,7 +314,7 @@
      * @see ChannelListener
      */
     public void removeChannelListener(ChannelListener listener);
-    
+
     /**
      * Returns true if there are any members in the group,
      * this call is the same as <code>getMembers().length>0</code>
@@ -317,7 +324,7 @@
 
     /**
      * Get all current group members
-     * @return all members or empty array, never null 
+     * @return all members or empty array, never null
      */
     public Member[] getMembers() ;
 
@@ -329,10 +336,10 @@
      * @return Member
      */
     public Member getLocalMember(boolean incAlive);
-    
+
     /**
-     * Returns the member from the membership service with complete and 
-     * recent data. Some implementations might serialize and send 
+     * Returns the member from the membership service with complete and
+     * recent data. Some implementations might serialize and send
      * membership information along with a message, and instead of sending
      * complete membership details, only send the primary identifier for the member
      * but not the payload or other information. When such message is received
@@ -343,5 +350,5 @@
      */
     public Member getMember(Member mbr);
 
-    
+
 }

Modified: tomcat/trunk/java/org/apache/catalina/tribes/io/ObjectReader.java
URL: http://svn.apache.org/viewvc/tomcat/trunk/java/org/apache/catalina/tribes/io/ObjectReader.java?rev=628881&r1=628880&r2=628881&view=diff
==============================================================================
--- tomcat/trunk/java/org/apache/catalina/tribes/io/ObjectReader.java (original)
+++ tomcat/trunk/java/org/apache/catalina/tribes/io/ObjectReader.java Mon Feb 18 14:07:09
2008
@@ -5,9 +5,9 @@
  * 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.
@@ -40,12 +40,15 @@
     protected static org.apache.juli.logging.Log log = org.apache.juli.logging.LogFactory.getLog(ObjectReader.class);
 
     private XByteBuffer buffer;
-    
+
     protected long lastAccess = System.currentTimeMillis();
-    
+
     protected boolean accessed = false;
     private boolean cancelled;
 
+    public ObjectReader(int packetSize) {
+        this.buffer = new XByteBuffer(packetSize, true);
+    }
     /**
      * Creates an <code>ObjectReader</code> for a TCP NIO socket channel
      * @param channel - the channel to be read.
@@ -53,7 +56,7 @@
     public ObjectReader(SocketChannel channel) {
         this(channel.socket());
     }
-    
+
     /**
      * Creates an <code>ObjectReader</code> for a TCP socket
      * @param socket Socket
@@ -67,23 +70,23 @@
             this.buffer = new XByteBuffer(43800,true);
         }
     }
-    
+
     public synchronized void access() {
         this.accessed = true;
         this.lastAccess = System.currentTimeMillis();
     }
-    
+
     public synchronized void finish() {
         this.accessed = false;
         this.lastAccess = System.currentTimeMillis();
     }
-    
+
     public boolean isAccessed() {
         return this.accessed;
     }
 
     /**
-     * Append new bytes to buffer. 
+     * Append new bytes to buffer.
      * @see XByteBuffer#countPackages()
      * @param data new transfer buffer
      * @param off offset
@@ -125,11 +128,11 @@
         }
         return result;
     }
-    
+
     public int bufferSize() {
         return buffer.getLength();
     }
-    
+
 
     public boolean hasPackage() {
         return buffer.countPackages(true)>0;
@@ -141,7 +144,7 @@
     public int count() {
         return buffer.countPackages();
     }
-    
+
     public void close() {
         this.buffer = null;
     }

Modified: tomcat/trunk/java/org/apache/catalina/tribes/transport/AbstractSender.java
URL: http://svn.apache.org/viewvc/tomcat/trunk/java/org/apache/catalina/tribes/transport/AbstractSender.java?rev=628881&r1=628880&r2=628881&view=diff
==============================================================================
--- tomcat/trunk/java/org/apache/catalina/tribes/transport/AbstractSender.java (original)
+++ tomcat/trunk/java/org/apache/catalina/tribes/transport/AbstractSender.java Mon Feb 18
14:07:09 2008
@@ -34,7 +34,7 @@
  * @version 1.0
  */
 public abstract class AbstractSender implements DataSender {
-    
+
     private boolean connected = false;
     private int rxBufSize = 25188;
     private int txBufSize = 43800;
@@ -57,7 +57,9 @@
     private int soLingerTime = 3;
     private int soTrafficClass = 0x04 | 0x08 | 0x010;
     private boolean throwOnFailedAck = true;
-    
+    private boolean udpBased = false;
+    private int udpPort = -1;
+
     /**
      * transfers sender properties from one sender to another
      * @param from AbstractSender
@@ -82,13 +84,15 @@
         to.soLingerTime = from.soLingerTime;
         to.soTrafficClass = from.soTrafficClass;
         to.throwOnFailedAck = from.throwOnFailedAck;
-    }   
+        to.udpBased = from.udpBased;
+        to.udpPort = from.udpPort;
+    }
+
 
-    
     public AbstractSender() {
-        
+
     }
-    
+
     /**
      * connect
      *
@@ -117,11 +121,11 @@
         if ( disconnect ) disconnect();
         return disconnect;
     }
-    
+
     protected void setConnected(boolean connected){
         this.connected = connected;
     }
-    
+
     public boolean isConnected() {
         return connected;
     }
@@ -170,7 +174,7 @@
     public int getMaxRetryAttempts() {
         return maxRetryAttempts;
     }
-    
+
     public void setDirect(boolean direct) {
         setDirectBuffer(direct);
     }
@@ -182,7 +186,7 @@
     public boolean getDirect() {
         return getDirectBuffer();
     }
-    
+
     public boolean getDirectBuffer() {
         return this.directBuffer;
     }
@@ -304,6 +308,26 @@
 
     public void setAddress(InetAddress address) {
         this.address = address;
+    }
+
+
+    public boolean isUdpBased() {
+        return udpBased;
+    }
+
+
+    public void setUdpBased(boolean udpBased) {
+        this.udpBased = udpBased;
+    }
+
+
+    public int getUdpPort() {
+        return udpPort;
+    }
+
+
+    public void setUdpPort(int udpPort) {
+        this.udpPort = udpPort;
     }
 
 }

Modified: tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java
URL: http://svn.apache.org/viewvc/tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java?rev=628881&r1=628880&r2=628881&view=diff
==============================================================================
--- tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java (original)
+++ tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioReceiver.java Mon Feb 18
14:07:09 2008
@@ -18,7 +18,9 @@
 package org.apache.catalina.tribes.transport.nio;
 
 import java.io.IOException;
+import java.net.InetSocketAddress;
 import java.net.ServerSocket;
+import java.nio.channels.DatagramChannel;
 import java.nio.channels.SelectableChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
@@ -59,6 +61,7 @@
 
     private Selector selector = null;
     private ServerSocketChannel serverChannel = null;
+    private DatagramChannel datagramChannel = null;
 
     protected LinkedList events = new LinkedList();
 //    private Object interestOpsMutex = new Object();
@@ -110,7 +113,7 @@
             else throw new IOException(x.getMessage());
         }
     }
-    
+
     public AbstractRxTask createRxTask() {
         NioReplicationTask thread = new NioReplicationTask(this,this);
         thread.setUseBufferPool(this.getUseBufferPool());
@@ -118,9 +121,9 @@
         thread.setOptions(getWorkerThreadOptions());
         return thread;
     }
-    
-    
-    
+
+
+
     protected void bind() throws IOException {
         // allocate an unbound server socket channel
         serverChannel = ServerSocketChannel.open();
@@ -135,9 +138,22 @@
         serverChannel.configureBlocking(false);
         // register the ServerSocketChannel with the Selector
         serverChannel.register(selector, SelectionKey.OP_ACCEPT);
-        
+
+        //set up the datagram channel
+        if (this.getUdpPort()>0) {
+            datagramChannel = DatagramChannel.open();
+            datagramChannel.configureBlocking(false);
+            //bind to the address to avoid security checks
+            InetSocketAddress daddr = new InetSocketAddress(getBind(),getUdpPort());
+            //TODO should we auto increment the UDP port to avoid collisions?
+            //we could auto increment with the offset from the tcp listen port
+            datagramChannel.connect(daddr);
+        }
+
+
+
     }
-    
+
     public void addEvent(Runnable event) {
         if ( selector != null ) {
             synchronized (events) {
@@ -163,18 +179,18 @@
             events.clear();
         }
     }
-    
+
     public static void cancelledKey(SelectionKey key) {
         ObjectReader reader = (ObjectReader)key.attachment();
         if ( reader != null ) {
             reader.setCancelled(true);
             reader.finish();
         }
-        key.cancel(); 
+        key.cancel();
         key.attach(null);
         try { ((SocketChannel)key.channel()).socket().close(); } catch (IOException e) {
if (log.isDebugEnabled()) log.debug("", e); }
         try { key.channel().close(); } catch (IOException e) { if (log.isDebugEnabled())
log.debug("", e); }
-        
+
     }
     protected long lastCheck = System.currentTimeMillis();
     protected void socketTimeouts() {
@@ -202,7 +218,7 @@
                     if ( ka != null ) {
                         long delta = now - ka.getLastAccess();
                         if (delta > (long) getTimeout() && (!ka.isAccessed()))
{
-                            if (log.isWarnEnabled()) 
+                            if (log.isWarnEnabled())
                                 log.warn("Channel key is registered, but has had no interest
ops for the last "+getTimeout()+" ms. (cancelled:"+ka.isCancelled()+"):"+key+" last access:"+new
java.sql.Timestamp(ka.getLastAccess())+" Possible cause: all threads used, perform thread
dump");
                             ka.setLastAccess(now);
                             //key.interestOps(SelectionKey.OP_READ);
@@ -230,8 +246,12 @@
             log.warn("ServerSocketChannel already started");
             return;
         }
-        
+
         setListen(true);
+        if (selector!=null && datagramChannel!=null) {
+            ObjectReader oreader = new ObjectReader(1024*65);
+            registerChannel(selector,datagramChannel,SelectionKey.OP_READ,oreader);
+        }
 
         while (doListen() && selector != null) {
             // this may block for a long time, upon return the
@@ -302,10 +322,18 @@
 
         }
         serverChannel.close();
+        if (datagramChannel!=null) {
+            try {
+                datagramChannel.close();
+            }catch (Exception iox) {
+                if (log.isDebugEnabled()) log.debug("Unable to close datagram channel.",iox);
+            }
+            datagramChannel=null;
+        }
         closeSelector();
     }
 
-    
+
 
     /**
      * Close Selector.

Modified: tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java
URL: http://svn.apache.org/viewvc/tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java?rev=628881&r1=628880&r2=628881&view=diff
==============================================================================
--- tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java (original)
+++ tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioReplicationTask.java Mon
Feb 18 14:07:09 2008
@@ -5,9 +5,9 @@
  * 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.
@@ -18,6 +18,9 @@
 package org.apache.catalina.tribes.transport.nio;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 
@@ -42,15 +45,15 @@
  * serviceChannel() method stores the key reference in the thread object then
  * calls notify() to wake it up. When the channel has been drained, the worker
  * thread returns itself to its parent pool.
- * 
+ *
  * @author Filip Hanik
- * 
+ *
  * @version $Revision$, $Date$
  */
 public class NioReplicationTask extends AbstractRxTask {
-    
+
     private static org.apache.juli.logging.Log log = org.apache.juli.logging.LogFactory.getLog(
NioReplicationTask.class );
-    
+
     private ByteBuffer buffer = null;
     private SelectionKey key;
     private int rxBufSize;
@@ -62,7 +65,7 @@
     }
 
     // loop forever waiting for work to do
-    public synchronized void run() { 
+    public synchronized void run() {
         if ( buffer == null ) {
             if ( (getOptions() & OPTION_DIRECT_BUFFER) == OPTION_DIRECT_BUFFER) {
                 buffer = ByteBuffer.allocateDirect(getRxBufSize());
@@ -75,17 +78,17 @@
         if (key == null) {
             return;	// just in case
         }
-        if ( log.isTraceEnabled() ) 
+        if ( log.isTraceEnabled() )
             log.trace("Servicing key:"+key);
 
         try {
             ObjectReader reader = (ObjectReader)key.attachment();
             if ( reader == null ) {
-                if ( log.isTraceEnabled() ) 
+                if ( log.isTraceEnabled() )
                     log.trace("No object reader, cancelling:"+key);
                 cancelKey(key);
             } else {
-                if ( log.isTraceEnabled() ) 
+                if ( log.isTraceEnabled() )
                     log.trace("Draining channel:"+key);
 
                 drainChannel(key, reader);
@@ -102,7 +105,7 @@
             } else if ( log.isErrorEnabled() ) {
                 //this is a real error, log it.
                 log.error("Exception caught in TcpReplicationThread.drainChannel.",e);
-            } 
+            }
             cancelKey(key);
         } finally {
 
@@ -143,16 +146,16 @@
     protected void drainChannel (final SelectionKey key, ObjectReader reader) throws Exception
{
         reader.setLastAccess(System.currentTimeMillis());
         reader.access();
-        SocketChannel channel = (SocketChannel) key.channel();
+        ReadableByteChannel channel = (ReadableByteChannel) key.channel();
         int count;
         buffer.clear();			// make buffer empty
 
         // loop while data available, channel is non-blocking
         while ((count = channel.read (buffer)) > 0) {
             buffer.flip();		// make buffer readable
-            if ( buffer.hasArray() ) 
+            if ( buffer.hasArray() )
                 reader.append(buffer.array(),0,count,false);
-            else 
+            else
                 reader.append(buffer,count,false);
             buffer.clear();		// make buffer empty
             //do we have at least one package?
@@ -160,24 +163,24 @@
         }
 
         int pkgcnt = reader.count();
-        
+
         if (count < 0 && pkgcnt == 0 ) {
             //end of stream, and no more packages to process
             remoteEof(key);
             return;
         }
-        
+
         ChannelMessage[] msgs = pkgcnt == 0? ChannelData.EMPTY_DATA_ARRAY : reader.execute();
-        
+
         registerForRead(key,reader);//register to read new data, before we send it off to
avoid dead locks
-        
+
         for ( int i=0; i<msgs.length; i++ ) {
             /**
-             * Use send ack here if you want to ack the request to the remote 
+             * Use send ack here if you want to ack the request to the remote
              * server before completing the request
              * This is considered an asynchronized request
              */
-            if (ChannelData.sendAckAsync(msgs[i].getOptions())) sendAck(key,channel,Constants.ACK_COMMAND);
+            if (ChannelData.sendAckAsync(msgs[i].getOptions())) sendAck(key,(WritableByteChannel)channel,Constants.ACK_COMMAND);
             try {
                 if ( Logs.MESSAGES.isTraceEnabled() ) {
                     try {
@@ -187,24 +190,24 @@
                 //process the message
                 getCallback().messageDataReceived(msgs[i]);
                 /**
-                 * Use send ack here if you want the request to complete on this 
+                 * Use send ack here if you want the request to complete on this
                  * server before sending the ack to the remote server
                  * This is considered a synchronized request
                  */
-                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(key,channel,Constants.ACK_COMMAND);
+                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(key,(WritableByteChannel)channel,Constants.ACK_COMMAND);
             }catch ( RemoteProcessException e ) {
                 if ( log.isDebugEnabled() ) log.error("Processing of cluster message failed.",e);
-                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(key,channel,Constants.FAIL_ACK_COMMAND);
+                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(key,(WritableByteChannel)channel,Constants.FAIL_ACK_COMMAND);
             }catch ( Exception e ) {
                 log.error("Processing of cluster message failed.",e);
-                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(key,channel,Constants.FAIL_ACK_COMMAND);
+                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(key,(WritableByteChannel)channel,Constants.FAIL_ACK_COMMAND);
             }
             if ( getUseBufferPool() ) {
                 BufferPool.getBufferPool().returnBuffer(msgs[i].getMessage());
                 msgs[i].setMessage(null);
             }
-        }                        
-        
+        }
+
         if (count < 0) {
             remoteEof(key);
             return;
@@ -218,7 +221,7 @@
     }
 
     protected void registerForRead(final SelectionKey key, ObjectReader reader) {
-        if ( log.isTraceEnabled() ) 
+        if ( log.isTraceEnabled() )
             log.trace("Adding key for read event:"+key);
         reader.finish();
         //register our OP_READ interest
@@ -231,12 +234,12 @@
                         // resume interest in OP_READ, OP_WRITE
                         int resumeOps = key.interestOps() | SelectionKey.OP_READ;
                         key.interestOps(resumeOps);
-                        if ( log.isTraceEnabled() ) 
+                        if ( log.isTraceEnabled() )
                             log.trace("Registering key for read:"+key);
                     }
                 } catch (CancelledKeyException ckx ) {
                     NioReceiver.cancelledKey(key);
-                    if ( log.isTraceEnabled() ) 
+                    if ( log.isTraceEnabled() )
                         log.trace("CKX Cancelling key:"+key);
 
                 } catch (Exception x) {
@@ -248,7 +251,7 @@
     }
 
     private void cancelKey(final SelectionKey key) {
-        if ( log.isTraceEnabled() ) 
+        if ( log.isTraceEnabled() )
             log.trace("Adding key for cancel event:"+key);
 
         ObjectReader reader = (ObjectReader)key.attachment();
@@ -258,7 +261,7 @@
         }
         Runnable cx = new Runnable() {
             public void run() {
-                if ( log.isTraceEnabled() ) 
+                if ( log.isTraceEnabled() )
                     log.trace("Cancelling key:"+key);
 
                 NioReceiver.cancelledKey(key);
@@ -266,8 +269,8 @@
         };
         receiver.addEvent(cx);
     }
-    
-    
+
+
 
 
 
@@ -276,8 +279,8 @@
      * @param key
      * @param channel
      */
-    protected void sendAck(SelectionKey key, SocketChannel channel, byte[] command) {
-        
+    protected void sendAck(SelectionKey key, WritableByteChannel channel, byte[] command)
{
+
         try {
             ByteBuffer buf = ByteBuffer.wrap(command);
             int total = 0;
@@ -285,7 +288,10 @@
                 total += channel.write(buf);
             }
             if (log.isTraceEnabled()) {
-                log.trace("ACK sent to " + channel.socket().getPort());
+                log.trace("ACK sent to " +
+                        ( (channel instanceof SocketChannel) ?
+                          ((SocketChannel)channel).socket().getInetAddress() :
+                          ((DatagramChannel)channel).socket().getInetAddress()));
             }
         } catch ( java.io.IOException x ) {
             log.warn("Unable to send ACK back through channel, channel disconnected?: "+x.getMessage());

Modified: tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioSender.java
URL: http://svn.apache.org/viewvc/tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioSender.java?rev=628881&r1=628880&r2=628881&view=diff
==============================================================================
--- tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioSender.java (original)
+++ tomcat/trunk/java/org/apache/catalina/tribes/transport/nio/NioSender.java Mon Feb 18 14:07:09
2008
@@ -5,9 +5,9 @@
  * 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.
@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
+import java.nio.channels.DatagramChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
@@ -34,7 +35,7 @@
 
 /**
  * This class is NOT thread safe and should never be used with more than one thread at a
time
- * 
+ *
  * This is a state machine, handled by the process method
  * States are:
  * - NOT_CONNECTED -> connect() -> CONNECTED
@@ -42,7 +43,7 @@
  * - READY_TO_WRITE -> write() -> READY TO WRITE | READY TO READ
  * - READY_TO_READ -> read() -> READY_TO_READ | TRANSFER_COMPLETE
  * - TRANSFER_COMPLETE -> CONNECTED
- * 
+ *
  * @author Filip Hanik
  * @version 1.0
  */
@@ -50,10 +51,11 @@
 
     protected static org.apache.juli.logging.Log log = org.apache.juli.logging.LogFactory.getLog(NioSender.class);
 
-    
-    
-    protected Selector selector;    
-    protected SocketChannel socketChannel;
+
+
+    protected Selector selector;
+    protected SocketChannel socketChannel = null;
+    protected DatagramChannel dataChannel = null;
 
     /*
      * STATE VARIABLES *
@@ -64,14 +66,14 @@
     protected XByteBuffer ackbuf = new XByteBuffer(128,true);
     protected int remaining = 0;
     protected boolean complete;
-    
+
     protected boolean connecting = false;
-    
+
     public NioSender() {
         super();
-        
+
     }
-    
+
     /**
      * State machine to send data
      * @param key SelectionKey
@@ -89,7 +91,7 @@
                 completeConnect();
                 if ( current != null ) key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
                 return false;
-            } else  { 
+            } else  {
                 //wait for the connection to finish
                 key.interestOps(key.interestOps() | SelectionKey.OP_CONNECT);
                 return false;
@@ -146,8 +148,8 @@
         socketChannel.socket().setSoLinger(getSoLingerOn(),getSoLingerTime());
         socketChannel.socket().setTrafficClass(getSoTrafficClass());
     }
-    
-    
+
+
 
     protected boolean read(SelectionKey key) throws IOException {
         //if there is no message here, we are done
@@ -171,7 +173,7 @@
         }
     }
 
-    
+
     protected boolean write(SelectionKey key) throws IOException {
         if ( (!isConnected()) || (this.socketChannel==null)) {
             throw new IOException("NioSender is not connected, this should not occur.");
@@ -215,7 +217,7 @@
         } else {
             writebuf.clear();
         }
-        
+
         InetSocketAddress addr = new InetSocketAddress(getAddress(),getPort());
         if ( socketChannel != null ) throw new IOException("Socket channel has already been
established. Connection might be in progress.");
         socketChannel = SocketChannel.open();
@@ -227,7 +229,7 @@
             socketChannel.register(getSelector(), SelectionKey.OP_CONNECT, this);
         }
     }
-    
+
 
     /**
      * disconnect
@@ -257,7 +259,7 @@
         }
 
     }
-    
+
     public void reset() {
         if ( isConnected() && readbuf == null) {
             readbuf = getReadBuffer();
@@ -273,10 +275,10 @@
         setConnectTime(-1);
     }
 
-    private ByteBuffer getReadBuffer() { 
+    private ByteBuffer getReadBuffer() {
         return getBuffer(getRxBufSize());
     }
-    
+
     private ByteBuffer getWriteBuffer() {
         return getBuffer(getTxBufSize());
     }
@@ -284,7 +286,7 @@
     private ByteBuffer getBuffer(int size) {
         return (getDirectBuffer()?ByteBuffer.allocateDirect(size):ByteBuffer.allocate(size));
     }
-    
+
     /**
     * sendMessage
     *
@@ -312,9 +314,9 @@
            if (isConnected()) {
                socketChannel.register(getSelector(), SelectionKey.OP_WRITE, this);
            }
-       } 
+       }
    }
-   
+
    public byte[] getMessage() {
        return current;
    }

Modified: tomcat/trunk/test/org/apache/catalina/tribes/test/channel/ChannelStartStop.java
URL: http://svn.apache.org/viewvc/tomcat/trunk/test/org/apache/catalina/tribes/test/channel/ChannelStartStop.java?rev=628881&r1=628880&r2=628881&view=diff
==============================================================================
--- tomcat/trunk/test/org/apache/catalina/tribes/test/channel/ChannelStartStop.java (original)
+++ tomcat/trunk/test/org/apache/catalina/tribes/test/channel/ChannelStartStop.java Mon Feb
18 14:07:09 2008
@@ -5,9 +5,9 @@
  * 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.
@@ -15,6 +15,7 @@
  */
 package org.apache.catalina.tribes.test.channel;
 
+import org.apache.catalina.tribes.Channel;
 import org.apache.catalina.tribes.group.GroupChannel;
 import junit.framework.TestCase;
 import org.apache.catalina.tribes.transport.ReceiverBase;
@@ -25,6 +26,7 @@
  */
 public class ChannelStartStop extends TestCase {
     GroupChannel channel = null;
+    int udpPort = 45543;
     protected void setUp() throws Exception {
         super.setUp();
         channel = new GroupChannel();
@@ -34,7 +36,7 @@
         super.tearDown();
         try {channel.stop(channel.DEFAULT);}catch (Exception ignore){}
     }
-    
+
     public void testDoubleFullStart() throws Exception {
         int count = 0;
         try {
@@ -52,11 +54,11 @@
     public void testScrap() throws Exception {
         System.out.println(channel.getChannelReceiver().getClass());
         ((ReceiverBase)channel.getChannelReceiver()).setMaxThreads(1);
-    } 
+    }
 
 
     public void testDoublePartialStart() throws Exception {
-        //try to double start the RX 
+        //try to double start the RX
         int count = 0;
         try {
             channel.start(channel.SND_RX_SEQ);
@@ -82,7 +84,7 @@
         } catch ( Exception x){/*expected*/}
         assertEquals(count,1);
         channel.stop(channel.DEFAULT);
-        
+
         count = 0;
         try {
             channel.start(channel.SND_RX_SEQ);
@@ -107,7 +109,7 @@
         assertEquals(count,1);
         channel.stop(channel.DEFAULT);
     }
-    
+
     public void testFalseOption() throws Exception {
         int flag = 0xFFF0;//should get ignored by the underlying components
         int count = 0;
@@ -121,6 +123,14 @@
         } catch ( Exception x){/*expected*/}
         assertEquals(count,2);
         channel.stop(channel.DEFAULT);
+    }
+
+    public void testUdpReceiverStart() throws Exception {
+        ReceiverBase rb = (ReceiverBase)channel.getChannelReceiver();
+        rb.setUdpPort(udpPort);
+        channel.start(Channel.DEFAULT);
+        Thread.sleep(1000);
+        channel.stop(Channel.DEFAULT);
     }
 
 }



---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org


Mime
View raw message