Coverage Report

Coverage Report - com.sun.grizzly.UDPConnectorHandler

Classes in this File

Line Coverage

Branch Coverage

Complexity

UDPConnectorHandler

47 %

66/139

36 %

21/58

UDPConnectorHandler$1

0 %

0/5

N/A

 /*
  * 
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  * 
  * Copyright 2007-2008 Sun Microsystems, Inc. All rights reserved.
  * 
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
  * and Distribution License("CDDL") (collectively, the "License").  You
  * may not use this file except in compliance with the License. You can obtain
  * a copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html
  * or glassfish/bootstrap/legal/LICENSE.txt.  See the License for the specific
  * language governing permissions and limitations under the License.
  * 
  * When distributing the software, include this License Header Notice in each
  * file and include the License file at glassfish/bootstrap/legal/LICENSE.txt.
  * Sun designates this particular file as subject to the "Classpath" exception
  * as provided by Sun in the GPL Version 2 section of the License file that
  * accompanied this code.  If applicable, add the following below the License
  * Header, with the fields enclosed by brackets [] replaced by your own
  * identifying information: "Portions Copyrighted [year]
  * [name of copyright owner]"
  * 
  * Contributor(s):
  * 
  * If you wish your version of this file to be governed by only the CDDL or
  * only the GPL Version 2, indicate your decision by adding "[Contributor]
  * elects to include this software in this distribution under the [CDDL or GPL
  * Version 2] license."  If you don't indicate a single choice of license, a
  * recipient has the option to distribute your version of this file under
  * either the CDDL, the GPL Version 2 or to extend the choice of license to
  * its licensees as provided above.  However, if you add GPL Version 2 code
  * and therefore, elected the GPL Version 2 license, then the option applies
  * only if the new code is made subject to such option by the copyright
  * holder.
  *
  */
 
 package com.sun.grizzly;
 
 import com.sun.grizzly.async.AsyncQueueDataProcessor;
 import com.sun.grizzly.async.AsyncQueueReadable;
 import com.sun.grizzly.async.AsyncWriteCallbackHandler;
 import com.sun.grizzly.async.AsyncQueueWritable;
 import com.sun.grizzly.async.AsyncReadCallbackHandler;
 import com.sun.grizzly.async.AsyncReadCondition;
 import com.sun.grizzly.util.InputReader;
 import java.io.IOException;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.nio.channels.AlreadyConnectedException;
 import java.nio.channels.NotYetConnectedException;
 import java.nio.channels.SelectableChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.DatagramChannel;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 
 /**
  * Client side interface used to implement non blocking client operation.
  * Implementation of this class must make sure the following methods are
  * invoked in that order:
  * <p><pre><code>
  * (1) connect()
  * (2) read() or write().
  * </code></pre></p>
  *
  * @author Jeanfrancois Arcand
  */
 public class UDPConnectorHandler implements 
         ConnectorHandler<UDPSelectorHandler, CallbackHandler>, 
         AsyncQueueWritable, AsyncQueueReadable {
     
     /**
      * The underlying UDPSelectorHandler used to mange SelectionKeys.
      */
     protected UDPSelectorHandler selectorHandler;
     
     
     /**
      * A {@link CallbackHandler} handler invoked by the UDPSelectorHandler
      * when a non blocking operation is ready to be processed.
      */
     protected CallbackHandler callbackHandler;
     
     
     /**
      * The connection's DatagramChannel.
      */
     protected DatagramChannel datagramChannel;
     
     
     /**
      * Is the connection established.
      */
     protected volatile boolean isConnected;
     
     
     /**
      * The internal Controller used (in case not specified).
      */
     protected Controller controller;
     
     
     /**
      * IsConnected Latch related
      */
     protected CountDownLatch isConnectedLatch;
     
     
     /**
      * Are we creating a controller every run.
      */
     private boolean isStandalone = false;
     
     
     /**
      * A blocking {@link InputStream} that use a pool of Selector
      * to execute a blocking read operation.
      */
     protected InputReader inputStream;
     
     
     /**
      * Connect to hostname:port. When an aysnchronous event happens (e.g
      * OP_READ or OP_WRITE), the {@link Controller} will invoke
      * the CallBackHandler.
      * @param remoteAddress remote address to connect
      * @param callbackHandler the handler invoked by its associated {@link SelectorHandler} when
      *        a non blocking operation is ready to be handled. When null, all 
      *        read and write operation will be delegated to the default
      *        {@link ProtocolChain} and its list of {@link ProtocolFilter} 
      *        . When null, this {@link ConnectorHandler} will create an instance of {@link DefaultCallbackHandler}.
      */
     public void connect(SocketAddress remoteAddress,
             CallbackHandler callbackHandler) throws IOException {
         
         connect(remoteAddress,null,callbackHandler);
     }
     
     
     /**
      * Connect to hostname:port. When an aysnchronous event happens (e.g
      * OP_READ or OP_WRITE), the {@link Controller} will invoke
      * the CallBackHandler.
      * @param remoteAddress remote address to connect
      * @param localAddress local address to bind
      * @param callbackHandler the handler invoked by its associated {@link SelectorHandler} when
      *        a non blocking operation is ready to be handled. When null, all 
      *        read and write operation will be delegated to the default
      *        {@link ProtocolChain} and its list of {@link ProtocolFilter} 
      *        . When null, this {@link ConnectorHandler} will create an instance of {@link DefaultCallbackHandler}.
      */
     public void connect(SocketAddress remoteAddress, SocketAddress localAddress,
             CallbackHandler callbackHandler) throws IOException {
         
         if (controller == null){
             throw new IllegalStateException("Controller cannot be null");
         }
         
         connect(remoteAddress,localAddress,callbackHandler,
                 (UDPSelectorHandler)controller.getSelectorHandler(protocol()));
     }
     
     
     /**
      * Connect to hostname:port. When an aysnchronous event happens (e.g
      * OP_READ or OP_WRITE), the {@link Controller} will invoke
      * the CallBackHandler.
      * @param remoteAddress remote address to connect
      * @param callbackHandler the handler invoked by its associated {@link SelectorHandler} when
      *        a non blocking operation is ready to be handled. When null, all 
      *        read and write operation will be delegated to the default
      *        {@link ProtocolChain} and its list of {@link ProtocolFilter} 
      *        . When null, this {@link ConnectorHandler} will create an instance of {@link DefaultCallbackHandler}.
      * @param selectorHandler an instance of SelectorHandler.
      */
     public void connect(SocketAddress remoteAddress,
             CallbackHandler callbackHandler,
             UDPSelectorHandler selectorHandler) throws IOException {
         
         connect(remoteAddress,null,callbackHandler,selectorHandler);
     }
     
     /**
      * Connect to hostname:port. When an aysnchronous event happens (e.g
      * OP_READ or OP_WRITE), the {@link Controller} will invoke
      * the CallBackHandler.
      * @param remoteAddress remote address to connect
      * @param localAddress local address to bin
      * @param callbackHandler the handler invoked by its associated {@link SelectorHandler} when
      *        a non blocking operation is ready to be handled. When null, all 
      *        read and write operation will be delegated to the default
      *        {@link ProtocolChain} and its list of {@link ProtocolFilter} 
      *        . When null, this {@link ConnectorHandler} will create an instance of {@link DefaultCallbackHandler}.
      * @param selectorHandler an instance of SelectorHandler.
      */
     public void connect(SocketAddress remoteAddress, SocketAddress localAddress,
             CallbackHandler callbackHandler,
             UDPSelectorHandler selectorHandler) throws IOException {
         
         if (isConnected){
             throw new AlreadyConnectedException();
         }
         
         if (controller == null){
             throw new IllegalStateException("Controller cannot be null");
         }
         
         if (selectorHandler == null){
             throw new IllegalStateException("Controller cannot be null");
         }
         
         this.selectorHandler = selectorHandler;
         if (callbackHandler == null){
             callbackHandler = new DefaultCallbackHandler(this);
         } else {
             this.callbackHandler = callbackHandler;
         }
         
         // Wait for the onConnect to be invoked.
         isConnectedLatch = new CountDownLatch(1);
         
         selectorHandler.connect(remoteAddress,localAddress,callbackHandler);
         
         try {
             isConnectedLatch.await(30, TimeUnit.SECONDS);
         } catch (InterruptedException ex) {
             throw new IOException(ex.getMessage());
         }
     }
     
     
     /**
      * Connect to hostname:port. Internally an instance of Controller and
      * its default SelectorHandler will be created everytime this method is
      * called. This method should be used only and only if no external
      * Controller has been initialized.
      * @param remoteAddress remote address to connect
      */
     public void connect(SocketAddress remoteAddress) throws IOException {
         connect(remoteAddress,(SocketAddress)null);
     }
     
     
     /**
      * Connect to hostname:port. Internally an instance of Controller and
      * its default SelectorHandler will be created everytime this method is
      * called. This method should be used only and only if no external
      * Controller has been initialized.
      * @param remoteAddress remote address to connect
      * @param localAddress local address to bin
      */
     public void connect(SocketAddress remoteAddress, SocketAddress localAddress)
             throws IOException {
         
         if (isConnected){
             throw new AlreadyConnectedException();
         }
         
         if (controller == null){
             isStandalone = true;
             controller = new Controller();
             controller.setSelectorHandler(new UDPSelectorHandler(true));
             DefaultPipeline pipeline = new DefaultPipeline();
             pipeline.initPipeline();
             pipeline.startPipeline();
             controller.setPipeline(pipeline);
             
             final CountDownLatch latch = new CountDownLatch(1);
             controller.addStateListener(new ControllerStateListenerAdapter() {
                 @Override
                 public void onReady() {
                     latch.countDown();
                 }
                 
                 @Override
                 public void onException(Throwable e) {
                     latch.countDown();
                 }
             });                     
             callbackHandler = new DefaultCallbackHandler(this, false);
             new Thread(controller, "GrizzlyUDPConnectorHandler-Controller").start();
             
             try {
                 latch.await();
             } catch (InterruptedException ex) {
             }
         }
         
         connect(remoteAddress,localAddress,callbackHandler,
                 (UDPSelectorHandler)controller.getSelectorHandler(protocol()));
     }
     
     
     /**
      * Read bytes. If blocking is set to <tt>true</tt>, a pool of temporary
      * {@link Selector} will be used to read bytes.
      * @param byteBuffer The byteBuffer to store bytes.
      * @param blocking <tt>true</tt> if a a pool of temporary Selector
      *        is required to handle a blocking read.
      */
     public long read(ByteBuffer byteBuffer, boolean blocking) throws IOException {
         if (!isConnected){
             throw new NotYetConnectedException();
         }
         
         SelectionKey key = datagramChannel.keyFor(selectorHandler.getSelector());
         if (blocking){
             if (inputStream == null) {
                 inputStream = new InputReader();
             }
             inputStream.setSelectionKey(key);
             inputStream.setChannelType(
                     InputReader.ChannelType.DatagramChannel);
             int nRead = inputStream.read(byteBuffer);
             return nRead;
         } else {
             if (callbackHandler == null){
                 throw new IllegalStateException
                         ("Non blocking read needs a CallbackHandler");
             }
             int nRead = datagramChannel.read(byteBuffer);
             
             if (nRead == 0){
                 selectorHandler.register(key, SelectionKey.OP_READ);
             }
             return nRead;
         }
     }
     
     
     /**
      * Writes bytes. If blocking is set to <tt>true</tt>, a pool of temporary
      * {@link Selector} will be used to writes bytes.
      * @param byteBuffer The byteBuffer to write.
      * @param blocking <tt>true</tt> if a a pool of temporary Selector
      *        is required to handle a blocking write.
      */
     public long write(ByteBuffer byteBuffer, boolean blocking) throws IOException {
         if (!isConnected){
             throw new NotYetConnectedException();
         }
                 
         if (blocking){
             throw new IllegalStateException("Blocking mode not supported");
         } else {
             if (callbackHandler == null){
                 throw new IllegalStateException
                         ("Non blocking write needs a CallbackHandler");
             }
             SelectionKey key = datagramChannel.keyFor(selectorHandler.getSelector());
             int nWrite = datagramChannel.write(byteBuffer);
             
             if (nWrite == 0){
                 selectorHandler.register(key, SelectionKey.OP_WRITE);
             }
             return nWrite;
         }
     }
     
     
     /**
      * {@inheritDoc}
      */
     public void readFromAsyncQueue(ByteBuffer buffer, 
             AsyncReadCallbackHandler callbackHandler) throws IOException {
         readFromAsyncQueue(buffer, callbackHandler, null);
     }
 
     /**
      * {@inheritDoc}
      */
     public void readFromAsyncQueue(ByteBuffer buffer, 
             AsyncReadCallbackHandler callbackHandler, 
             AsyncReadCondition condition) throws IOException {
         readFromAsyncQueue(buffer, callbackHandler, condition, null);
     }
 
     /**
      * {@inheritDoc}
      */
     public void readFromAsyncQueue(ByteBuffer buffer, 
             AsyncReadCallbackHandler callbackHandler, 
             AsyncReadCondition condition, 
             AsyncQueueDataProcessor readPostProcessor) throws IOException {
         selectorHandler.getAsyncQueueReader().read(
                 datagramChannel.keyFor(selectorHandler.getSelector()), buffer, 
                 callbackHandler, condition, readPostProcessor);
     }
 
     /**
      * {@inheritDoc}
      */
     public void writeToAsyncQueue(ByteBuffer buffer) throws IOException {
         writeToAsyncQueue(buffer, null);
     }
 
 
     /**
      * {@inheritDoc}
      */
     public void writeToAsyncQueue(ByteBuffer buffer, 
             AsyncWriteCallbackHandler callbackHandler) throws IOException {
         writeToAsyncQueue(buffer, callbackHandler, null);
     }
 
 
     /**
      * {@inheritDoc}
      */
     public void writeToAsyncQueue(ByteBuffer buffer, 
             AsyncWriteCallbackHandler callbackHandler,
             AsyncQueueDataProcessor writePreProcessor) throws IOException {
         writeToAsyncQueue(buffer, callbackHandler, writePreProcessor, false);
     }
 
     
     /**
      * {@inheritDoc}
      */
     public void writeToAsyncQueue(ByteBuffer buffer, 
             AsyncWriteCallbackHandler callbackHandler, 
             AsyncQueueDataProcessor writePreProcessor,
             boolean isCloneByteBuffer) throws IOException {
         selectorHandler.getAsyncQueueWriter().write(
                 datagramChannel.keyFor(selectorHandler.getSelector()), buffer,
                 callbackHandler, writePreProcessor, isCloneByteBuffer);
     }
 
     
     /**
      * {@inheritDoc}
      */
     public void writeToAsyncQueue(SocketAddress dstAddress, ByteBuffer buffer)
             throws IOException {
         writeToAsyncQueue(dstAddress, buffer, null);
     }
 
 
     /**
      * {@inheritDoc}
      */
     public void writeToAsyncQueue(SocketAddress dstAddress, ByteBuffer buffer, 
             AsyncWriteCallbackHandler callbackHandler) throws IOException {
         writeToAsyncQueue(dstAddress, buffer, callbackHandler, null);
     }
 
 
     /**
      * {@inheritDoc}
      */
     public void writeToAsyncQueue(SocketAddress dstAddress, ByteBuffer buffer, 
             AsyncWriteCallbackHandler callbackHandler,
             AsyncQueueDataProcessor writePreProcessor) throws IOException {
         writeToAsyncQueue(dstAddress, buffer, callbackHandler, writePreProcessor,
                 false);
     }
 
     
     /**
      * {@inheritDoc}
      */
     public void writeToAsyncQueue(SocketAddress dstAddress, ByteBuffer buffer, 
             AsyncWriteCallbackHandler callbackHandler, 
             AsyncQueueDataProcessor writePreProcessor, boolean isCloneByteBuffer)
             throws IOException {
         selectorHandler.getAsyncQueueWriter().write(
                 datagramChannel.keyFor(selectorHandler.getSelector()), dstAddress,
                 buffer, callbackHandler, writePreProcessor, isCloneByteBuffer);
     }
 
     
     /**
      * Receive bytes.
      * @param byteBuffer The byteBuffer to store bytes.
      * @param socketAddress
      * @return number bytes sent
      * @throws java.io.IOException
      */
     public long send(ByteBuffer byteBuffer, SocketAddress socketAddress)
     throws IOException {
         if (!isConnected){
             throw new NotYetConnectedException();
         }
         
         if (callbackHandler == null){
             throw new IllegalStateException
                     ("Non blocking read needs a CallbackHandler");
         }
         
         return datagramChannel.send(byteBuffer,socketAddress);
     }
     
     
     /**
      * Receive bytes.
      * @param byteBuffer The byteBuffer to store bytes.
      * @return {@link SocketAddress}
      * @throws java.io.IOException
      */
     public SocketAddress receive(ByteBuffer byteBuffer) throws IOException {
         if (!isConnected){
             throw new NotYetConnectedException();
         }
         
         SelectionKey key = datagramChannel.keyFor(selectorHandler.getSelector());
         
         if (callbackHandler == null){
             throw new IllegalStateException
                     ("Non blocking read needs a CallbackHandler");
         }
         
         SocketAddress socketAddress = datagramChannel.receive(byteBuffer);
         return socketAddress;
     }
     
     
     /**
      * Close the underlying connection.
      */
     public void close() throws IOException{
         if (datagramChannel != null){
             if (selectorHandler != null){
                 SelectionKey key =
                         datagramChannel.keyFor(selectorHandler.getSelector());
                 
                 if (key == null) return;
                 
                 selectorHandler.getSelectionKeyHandler().cancel(key);
             } else {
                 datagramChannel.close();
             }
         }
         
         if (controller != null && isStandalone){
             controller.stop();
             controller = null;
         }
         
         isStandalone = false;
         isConnected = false;
     }
     
     
     /**
      * Finish handling the OP_CONNECT interest ops.
      */
     public void finishConnect(SelectionKey key) throws IOException {
         if (Controller.logger().isLoggable(Level.FINE)) {
             Controller.logger().log(Level.FINE, "Finish connect");
         }
         
         datagramChannel = (DatagramChannel)key.channel();
         isConnected = datagramChannel.isConnected();
         if (isConnectedLatch != null) {
             isConnectedLatch.countDown();
         }
     }
     
     
     /**
      * A token decribing the protocol supported by an implementation of this
      * interface
      */
     public Controller.Protocol protocol(){
         return Controller.Protocol.UDP;
     }
     
     
     /**
      * Is the underlying DatagramChannel connected.
      * @return true if connected, othewise false
      */
     public boolean isConnected(){
         return isConnected && datagramChannel.isOpen();
     }
     
     
     /**
      * Return the  {@link Controller}
      * @return the  {@link Controller}
      */
     public Controller getController() {
         return controller;
     }
     
     
     /**
      * Set the {@link Controller} to use with this instance.
      * @param controller the {@link Controller} to use with this instance.
      */
     public void setController(Controller controller) {
         this.controller = controller;
     }
     
     
     /**
      * Return the current {@link SocketChannel} used.
      * @return the current {@link SocketChannel} used.
      */
     public SelectableChannel getUnderlyingChannel() {
         return datagramChannel;
     }
     
     
     /**
      * Return the {@link CallbackHandler}. 
      * @return the {@link CallbackHandler}. 
      */
     public CallbackHandler getCallbackHandler() {
         return callbackHandler;
     }
     
 
     /**
      * Set the {@link CallbackHandler}. 
      * @param callbackHandler the {@link CallbackHandler}. 
      */   
     public void setCallbackHandler(CallbackHandler callbackHandler) {
         this.callbackHandler = callbackHandler;
     }   
     
         
     /**
      * Return the associated {@link SelectorHandler}.
      * @return the associated {@link SelectorHandler}.
      */
     public UDPSelectorHandler getSelectorHandler() {
         return selectorHandler;
     }
 }

1		/*
2		*
3		* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
4		*
5		* Copyright 2007-2008 Sun Microsystems, Inc. All rights reserved.
6		*
7		* The contents of this file are subject to the terms of either the GNU
8		* General Public License Version 2 only ("GPL") or the Common Development
9		* and Distribution License("CDDL") (collectively, the "License"). You
10		* may not use this file except in compliance with the License. You can obtain
11		* a copy of the License at https://glassfish.dev.java.net/public/CDDL+GPL.html
12		* or glassfish/bootstrap/legal/LICENSE.txt. See the License for the specific
13		* language governing permissions and limitations under the License.
14		*
15		* When distributing the software, include this License Header Notice in each
16		* file and include the License file at glassfish/bootstrap/legal/LICENSE.txt.
17		* Sun designates this particular file as subject to the "Classpath" exception
18		* as provided by Sun in the GPL Version 2 section of the License file that
19		* accompanied this code. If applicable, add the following below the License
20		* Header, with the fields enclosed by brackets [] replaced by your own
21		* identifying information: "Portions Copyrighted [year]
22		* [name of copyright owner]"
23		*
24		* Contributor(s):
25		*
26		* If you wish your version of this file to be governed by only the CDDL or
27		* only the GPL Version 2, indicate your decision by adding "[Contributor]
28		* elects to include this software in this distribution under the [CDDL or GPL
29		* Version 2] license." If you don't indicate a single choice of license, a
30		* recipient has the option to distribute your version of this file under
31		* either the CDDL, the GPL Version 2 or to extend the choice of license to
32		* its licensees as provided above. However, if you add GPL Version 2 code
33		* and therefore, elected the GPL Version 2 license, then the option applies
34		* only if the new code is made subject to such option by the copyright
35		* holder.
36		*
37		*/
38
39		package com.sun.grizzly;
40
41		import com.sun.grizzly.async.AsyncQueueDataProcessor;
42		import com.sun.grizzly.async.AsyncQueueReadable;
43		import com.sun.grizzly.async.AsyncWriteCallbackHandler;
44		import com.sun.grizzly.async.AsyncQueueWritable;
45		import com.sun.grizzly.async.AsyncReadCallbackHandler;
46		import com.sun.grizzly.async.AsyncReadCondition;
47		import com.sun.grizzly.util.InputReader;
48		import java.io.IOException;
49		import java.net.SocketAddress;
50		import java.nio.ByteBuffer;
51		import java.nio.channels.AlreadyConnectedException;
52		import java.nio.channels.NotYetConnectedException;
53		import java.nio.channels.SelectableChannel;
54		import java.nio.channels.SelectionKey;
55		import java.nio.channels.DatagramChannel;
56		import java.util.concurrent.CountDownLatch;
57		import java.util.concurrent.TimeUnit;
58		import java.util.logging.Level;
59
60		/**
61		* Client side interface used to implement non blocking client operation.
62		* Implementation of this class must make sure the following methods are
63		* invoked in that order:
64		* <p><pre><code>
65		* (1) connect()
66		* (2) read() or write().
67		* </code></pre></p>
68		*
69		* @author Jeanfrancois Arcand
70		*/
71	4	public class UDPConnectorHandler implements
72		ConnectorHandler<UDPSelectorHandler, CallbackHandler>,
73		AsyncQueueWritable, AsyncQueueReadable {
74
75		/**
76		* The underlying UDPSelectorHandler used to mange SelectionKeys.
77		*/
78		protected UDPSelectorHandler selectorHandler;
79
80
81		/**
82		* A {@link CallbackHandler} handler invoked by the UDPSelectorHandler
83		* when a non blocking operation is ready to be processed.
84		*/
85		protected CallbackHandler callbackHandler;
86
87
88		/**
89		* The connection's DatagramChannel.
90		*/
91		protected DatagramChannel datagramChannel;
92
93
94		/**
95		* Is the connection established.
96		*/
97		protected volatile boolean isConnected;
98
99
100		/**
101		* The internal Controller used (in case not specified).
102		*/
103		protected Controller controller;
104
105
106		/**
107		* IsConnected Latch related
108		*/
109		protected CountDownLatch isConnectedLatch;
110
111
112		/**
113		* Are we creating a controller every run.
114		*/
115	4	private boolean isStandalone = false;
116
117
118		/**
119		* A blocking {@link InputStream} that use a pool of Selector
120		* to execute a blocking read operation.
121		*/
122		protected InputReader inputStream;
123
124
125		/**
126		* Connect to hostname:port. When an aysnchronous event happens (e.g
127		* OP_READ or OP_WRITE), the {@link Controller} will invoke
128		* the CallBackHandler.
129		* @param remoteAddress remote address to connect
130		* @param callbackHandler the handler invoked by its associated {@link SelectorHandler} when
131		* a non blocking operation is ready to be handled. When null, all
132		* read and write operation will be delegated to the default
133		* {@link ProtocolChain} and its list of {@link ProtocolFilter}
134		* . When null, this {@link ConnectorHandler} will create an instance of {@link DefaultCallbackHandler}.
135		*/
136		public void connect(SocketAddress remoteAddress,
137		CallbackHandler callbackHandler) throws IOException {
138
139	31	connect(remoteAddress,null,callbackHandler);
140	31	}
141
142
143		/**
144		* Connect to hostname:port. When an aysnchronous event happens (e.g
145		* OP_READ or OP_WRITE), the {@link Controller} will invoke
146		* the CallBackHandler.
147		* @param remoteAddress remote address to connect
148		* @param localAddress local address to bind
149		* @param callbackHandler the handler invoked by its associated {@link SelectorHandler} when
150		* a non blocking operation is ready to be handled. When null, all
151		* read and write operation will be delegated to the default
152		* {@link ProtocolChain} and its list of {@link ProtocolFilter}
153		* . When null, this {@link ConnectorHandler} will create an instance of {@link DefaultCallbackHandler}.
154		*/
155		public void connect(SocketAddress remoteAddress, SocketAddress localAddress,
156		CallbackHandler callbackHandler) throws IOException {
157
158	31	if (controller == null){
159	0	throw new IllegalStateException("Controller cannot be null");
160		}
161
162	31	connect(remoteAddress,localAddress,callbackHandler,
163		(UDPSelectorHandler)controller.getSelectorHandler(protocol()));
164	31	}
165
166
167		/**
168		* Connect to hostname:port. When an aysnchronous event happens (e.g
169		* OP_READ or OP_WRITE), the {@link Controller} will invoke
170		* the CallBackHandler.
171		* @param remoteAddress remote address to connect
172		* @param callbackHandler the handler invoked by its associated {@link SelectorHandler} when
173		* a non blocking operation is ready to be handled. When null, all
174		* read and write operation will be delegated to the default
175		* {@link ProtocolChain} and its list of {@link ProtocolFilter}
176		* . When null, this {@link ConnectorHandler} will create an instance of {@link DefaultCallbackHandler}.
177		* @param selectorHandler an instance of SelectorHandler.
178		*/
179		public void connect(SocketAddress remoteAddress,
180		CallbackHandler callbackHandler,
181		UDPSelectorHandler selectorHandler) throws IOException {
182
183	0	connect(remoteAddress,null,callbackHandler,selectorHandler);
184	0	}
185
186		/**
187		* Connect to hostname:port. When an aysnchronous event happens (e.g
188		* OP_READ or OP_WRITE), the {@link Controller} will invoke
189		* the CallBackHandler.
190		* @param remoteAddress remote address to connect
191		* @param localAddress local address to bin
192		* @param callbackHandler the handler invoked by its associated {@link SelectorHandler} when
193		* a non blocking operation is ready to be handled. When null, all
194		* read and write operation will be delegated to the default
195		* {@link ProtocolChain} and its list of {@link ProtocolFilter}
196		* . When null, this {@link ConnectorHandler} will create an instance of {@link DefaultCallbackHandler}.
197		* @param selectorHandler an instance of SelectorHandler.
198		*/
199		public void connect(SocketAddress remoteAddress, SocketAddress localAddress,
200		CallbackHandler callbackHandler,
201		UDPSelectorHandler selectorHandler) throws IOException {
202
203	31	if (isConnected){
204	0	throw new AlreadyConnectedException();
205		}
206
207	31	if (controller == null){
208	0	throw new IllegalStateException("Controller cannot be null");
209		}
210
211	31	if (selectorHandler == null){
212	0	throw new IllegalStateException("Controller cannot be null");
213		}
214
215	31	this.selectorHandler = selectorHandler;
216	31	if (callbackHandler == null){
217	0	callbackHandler = new DefaultCallbackHandler(this);
218		} else {
219	31	this.callbackHandler = callbackHandler;
220		}
221
222		// Wait for the onConnect to be invoked.
223	31	isConnectedLatch = new CountDownLatch(1);
224
225	31	selectorHandler.connect(remoteAddress,localAddress,callbackHandler);
226
227		try {
228	31	isConnectedLatch.await(30, TimeUnit.SECONDS);
229	0	} catch (InterruptedException ex) {
230	0	throw new IOException(ex.getMessage());
231	31	}
232	31	}
233
234
235		/**
236		* Connect to hostname:port. Internally an instance of Controller and
237		* its default SelectorHandler will be created everytime this method is
238		* called. This method should be used only and only if no external
239		* Controller has been initialized.
240		* @param remoteAddress remote address to connect
241		*/
242		public void connect(SocketAddress remoteAddress) throws IOException {
243	0	connect(remoteAddress,(SocketAddress)null);
244	0	}
245
246
247		/**
248		* Connect to hostname:port. Internally an instance of Controller and
249		* its default SelectorHandler will be created everytime this method is
250		* called. This method should be used only and only if no external
251		* Controller has been initialized.
252		* @param remoteAddress remote address to connect
253		* @param localAddress local address to bin
254		*/
255		public void connect(SocketAddress remoteAddress, SocketAddress localAddress)
256		throws IOException {
257
258	0	if (isConnected){
259	0	throw new AlreadyConnectedException();
260		}
261
262	0	if (controller == null){
263	0	isStandalone = true;
264	0	controller = new Controller();
265	0	controller.setSelectorHandler(new UDPSelectorHandler(true));
266	0	DefaultPipeline pipeline = new DefaultPipeline();
267	0	pipeline.initPipeline();
268	0	pipeline.startPipeline();
269	0	controller.setPipeline(pipeline);
270
271	0	final CountDownLatch latch = new CountDownLatch(1);
272	0	controller.addStateListener(new ControllerStateListenerAdapter() {
273		@Override
274		public void onReady() {
275	0	latch.countDown();
276	0	}
277
278		@Override
279		public void onException(Throwable e) {
280	0	latch.countDown();
281	0	}
282		});
283	0	callbackHandler = new DefaultCallbackHandler(this, false);
284	0	new Thread(controller, "GrizzlyUDPConnectorHandler-Controller").start();
285
286		try {
287	0	latch.await();
288	0	} catch (InterruptedException ex) {
289	0	}
290		}
291
292	0	connect(remoteAddress,localAddress,callbackHandler,
293		(UDPSelectorHandler)controller.getSelectorHandler(protocol()));
294	0	}
295
296
297		/**
298		* Read bytes. If blocking is set to <tt>true</tt>, a pool of temporary
299		* {@link Selector} will be used to read bytes.
300		* @param byteBuffer The byteBuffer to store bytes.
301		* @param blocking <tt>true</tt> if a a pool of temporary Selector
302		* is required to handle a blocking read.
303		*/
304		public long read(ByteBuffer byteBuffer, boolean blocking) throws IOException {
305	2250	if (!isConnected){
306	0	throw new NotYetConnectedException();
307		}
308
309	2250	SelectionKey key = datagramChannel.keyFor(selectorHandler.getSelector());
310	2250	if (blocking){
311	0	if (inputStream == null) {
312	0	inputStream = new InputReader();
313		}
314	0	inputStream.setSelectionKey(key);
315	0	inputStream.setChannelType(
316		InputReader.ChannelType.DatagramChannel);
317	0	int nRead = inputStream.read(byteBuffer);
318	0	return nRead;
319		} else {
320	2250	if (callbackHandler == null){
321	0	throw new IllegalStateException
322		("Non blocking read needs a CallbackHandler");
323		}
324	2250	int nRead = datagramChannel.read(byteBuffer);
325
326	2250	if (nRead == 0){
327	1085	selectorHandler.register(key, SelectionKey.OP_READ);
328		}
329	2250	return nRead;
330		}
331		}
332
333
334		/**
335		* Writes bytes. If blocking is set to <tt>true</tt>, a pool of temporary
336		* {@link Selector} will be used to writes bytes.
337		* @param byteBuffer The byteBuffer to write.
338		* @param blocking <tt>true</tt> if a a pool of temporary Selector
339		* is required to handle a blocking write.
340		*/
341		public long write(ByteBuffer byteBuffer, boolean blocking) throws IOException {
342	1001	if (!isConnected){
343	0	throw new NotYetConnectedException();
344		}
345
346	1001	if (blocking){
347	0	throw new IllegalStateException("Blocking mode not supported");
348		} else {
349	1001	if (callbackHandler == null){
350	0	throw new IllegalStateException
351		("Non blocking write needs a CallbackHandler");
352		}
353	1001	SelectionKey key = datagramChannel.keyFor(selectorHandler.getSelector());
354	1001	int nWrite = datagramChannel.write(byteBuffer);
355
356	1001	if (nWrite == 0){
357	0	selectorHandler.register(key, SelectionKey.OP_WRITE);
358		}
359	1001	return nWrite;
360		}
361		}
362
363
364		/**
365		* {@inheritDoc}
366		*/
367		public void readFromAsyncQueue(ByteBuffer buffer,
368		AsyncReadCallbackHandler callbackHandler) throws IOException {
369	100	readFromAsyncQueue(buffer, callbackHandler, null);
370	100	}
371
372		/**
373		* {@inheritDoc}
374		*/
375		public void readFromAsyncQueue(ByteBuffer buffer,
376		AsyncReadCallbackHandler callbackHandler,
377		AsyncReadCondition condition) throws IOException {
378	100	readFromAsyncQueue(buffer, callbackHandler, condition, null);
379	100	}
380
381		/**
382		* {@inheritDoc}
383		*/
384		public void readFromAsyncQueue(ByteBuffer buffer,
385		AsyncReadCallbackHandler callbackHandler,
386		AsyncReadCondition condition,
387		AsyncQueueDataProcessor readPostProcessor) throws IOException {
388	100	selectorHandler.getAsyncQueueReader().read(
389		datagramChannel.keyFor(selectorHandler.getSelector()), buffer,
390		callbackHandler, condition, readPostProcessor);
391	100	}
392
393		/**
394		* {@inheritDoc}
395		*/
396		public void writeToAsyncQueue(ByteBuffer buffer) throws IOException {
397	20000	writeToAsyncQueue(buffer, null);
398	20000	}
399
400
401		/**
402		* {@inheritDoc}
403		*/
404		public void writeToAsyncQueue(ByteBuffer buffer,
405		AsyncWriteCallbackHandler callbackHandler) throws IOException {
406	20000	writeToAsyncQueue(buffer, callbackHandler, null);
407	20000	}
408
409
410		/**
411		* {@inheritDoc}
412		*/
413		public void writeToAsyncQueue(ByteBuffer buffer,
414		AsyncWriteCallbackHandler callbackHandler,
415		AsyncQueueDataProcessor writePreProcessor) throws IOException {
416	20000	writeToAsyncQueue(buffer, callbackHandler, writePreProcessor, false);
417	20000	}
418
419
420		/**
421		* {@inheritDoc}
422		*/
423		public void writeToAsyncQueue(ByteBuffer buffer,
424		AsyncWriteCallbackHandler callbackHandler,
425		AsyncQueueDataProcessor writePreProcessor,
426		boolean isCloneByteBuffer) throws IOException {
427	20000	selectorHandler.getAsyncQueueWriter().write(
428		datagramChannel.keyFor(selectorHandler.getSelector()), buffer,
429		callbackHandler, writePreProcessor, isCloneByteBuffer);
430	20000	}
431
432
433		/**
434		* {@inheritDoc}
435		*/
436		public void writeToAsyncQueue(SocketAddress dstAddress, ByteBuffer buffer)
437		throws IOException {
438	0	writeToAsyncQueue(dstAddress, buffer, null);
439	0	}
440
441
442		/**
443		* {@inheritDoc}
444		*/
445		public void writeToAsyncQueue(SocketAddress dstAddress, ByteBuffer buffer,
446		AsyncWriteCallbackHandler callbackHandler) throws IOException {
447	0	writeToAsyncQueue(dstAddress, buffer, callbackHandler, null);
448	0	}
449
450
451		/**
452		* {@inheritDoc}
453		*/
454		public void writeToAsyncQueue(SocketAddress dstAddress, ByteBuffer buffer,
455		AsyncWriteCallbackHandler callbackHandler,
456		AsyncQueueDataProcessor writePreProcessor) throws IOException {
457	0	writeToAsyncQueue(dstAddress, buffer, callbackHandler, writePreProcessor,
458		false);
459	0	}
460
461
462		/**
463		* {@inheritDoc}
464		*/
465		public void writeToAsyncQueue(SocketAddress dstAddress, ByteBuffer buffer,
466		AsyncWriteCallbackHandler callbackHandler,
467		AsyncQueueDataProcessor writePreProcessor, boolean isCloneByteBuffer)
468		throws IOException {
469	0	selectorHandler.getAsyncQueueWriter().write(
470		datagramChannel.keyFor(selectorHandler.getSelector()), dstAddress,
471		buffer, callbackHandler, writePreProcessor, isCloneByteBuffer);
472	0	}
473
474
475		/**
476		* Receive bytes.
477		* @param byteBuffer The byteBuffer to store bytes.
478		* @param socketAddress
479		* @return number bytes sent
480		* @throws java.io.IOException
481		*/
482		public long send(ByteBuffer byteBuffer, SocketAddress socketAddress)
483		throws IOException {
484	0	if (!isConnected){
485	0	throw new NotYetConnectedException();
486		}
487
488	0	if (callbackHandler == null){
489	0	throw new IllegalStateException
490		("Non blocking read needs a CallbackHandler");
491		}
492
493	0	return datagramChannel.send(byteBuffer,socketAddress);
494		}
495
496
497		/**
498		* Receive bytes.
499		* @param byteBuffer The byteBuffer to store bytes.
500		* @return {@link SocketAddress}
501		* @throws java.io.IOException
502		*/
503		public SocketAddress receive(ByteBuffer byteBuffer) throws IOException {
504	0	if (!isConnected){
505	0	throw new NotYetConnectedException();
506		}
507
508	0	SelectionKey key = datagramChannel.keyFor(selectorHandler.getSelector());
509
510	0	if (callbackHandler == null){
511	0	throw new IllegalStateException
512		("Non blocking read needs a CallbackHandler");
513		}
514
515	0	SocketAddress socketAddress = datagramChannel.receive(byteBuffer);
516	0	return socketAddress;
517		}
518
519
520		/**
521		* Close the underlying connection.
522		*/
523		public void close() throws IOException{
524	31	if (datagramChannel != null){
525	31	if (selectorHandler != null){
526	31	SelectionKey key =
527		datagramChannel.keyFor(selectorHandler.getSelector());
528
529	31	if (key == null) return;
530
531	31	selectorHandler.getSelectionKeyHandler().cancel(key);
532	31	} else {
533	0	datagramChannel.close();
534		}
535		}
536
537	31	if (controller != null && isStandalone){
538	0	controller.stop();
539	0	controller = null;
540		}
541
542	31	isStandalone = false;
543	31	isConnected = false;
544	31	}
545
546
547		/**
548		* Finish handling the OP_CONNECT interest ops.
549		*/
550		public void finishConnect(SelectionKey key) throws IOException {
551	31	if (Controller.logger().isLoggable(Level.FINE)) {
552	0	Controller.logger().log(Level.FINE, "Finish connect");
553		}
554
555	31	datagramChannel = (DatagramChannel)key.channel();
556	31	isConnected = datagramChannel.isConnected();
557	31	if (isConnectedLatch != null) {
558	31	isConnectedLatch.countDown();
559		}
560	31	}
561
562
563		/**
564		* A token decribing the protocol supported by an implementation of this
565		* interface
566		*/
567		public Controller.Protocol protocol(){
568	62	return Controller.Protocol.UDP;
569		}
570
571
572		/**
573		* Is the underlying DatagramChannel connected.
574		* @return true if connected, othewise false
575		*/
576		public boolean isConnected(){
577	0	return isConnected && datagramChannel.isOpen();
578		}
579
580
581		/**
582		* Return the {@link Controller}
583		* @return the {@link Controller}
584		*/
585		public Controller getController() {
586	0	return controller;
587		}
588
589
590		/**
591		* Set the {@link Controller} to use with this instance.
592		* @param controller the {@link Controller} to use with this instance.
593		*/
594		public void setController(Controller controller) {
595	62	this.controller = controller;
596	62	}
597
598
599		/**
600		* Return the current {@link SocketChannel} used.
601		* @return the current {@link SocketChannel} used.
602		*/
603		public SelectableChannel getUnderlyingChannel() {
604	0	return datagramChannel;
605		}
606
607
608		/**
609		* Return the {@link CallbackHandler}.
610		* @return the {@link CallbackHandler}.
611		*/
612		public CallbackHandler getCallbackHandler() {
613	0	return callbackHandler;
614		}
615
616
617		/**
618		* Set the {@link CallbackHandler}.
619		* @param callbackHandler the {@link CallbackHandler}.
620		*/
621		public void setCallbackHandler(CallbackHandler callbackHandler) {
622	0	this.callbackHandler = callbackHandler;
623	0	}
624
625
626		/**
627		* Return the associated {@link SelectorHandler}.
628		* @return the associated {@link SelectorHandler}.
629		*/
630		public UDPSelectorHandler getSelectorHandler() {
631	0	return selectorHandler;
632		}
633		}