--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel;
+
+import java.util.Collection;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * Base implementation of the Kernel interface providing several
+ * useful default implementations of some methods. This implementation
+ * assumes that the kernel will be managing its own internal threads
+ * and queuing any results for the caller to retrieve on their own
+ * thread.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public abstract class AbstractKernel implements Kernel
+{
+ static Logger log = Logger.getLogger(AbstractKernel.class.getName());
+
+ private AtomicLong nextId = new AtomicLong(1);
+
+ /**
+ * Contains the pending endpoint events waiting for the caller
+ * to retrieve them.
+ */
+ private ConcurrentLinkedQueue<EndpointEvent> endpointEvents = new ConcurrentLinkedQueue<EndpointEvent>();
+
+ /**
+ * Contains the pending envelopes waiting for the caller to
+ * retrieve them.
+ */
+ private LinkedBlockingQueue<Envelope> envelopes = new LinkedBlockingQueue<Envelope>();
+
+ protected AbstractKernel()
+ {
+ }
+
+ protected void reportError( Exception e )
+ {
+ // Should really be queued up so the outer thread can
+ // retrieve them. For now we'll just log it. FIXME
+ log.log( Level.SEVERE, "Unhanddled kernel error", e );
+ }
+
+ protected long nextEndpointId()
+ {
+ return nextId.getAndIncrement();
+ }
+
+ /**
+ * Returns true if there are waiting envelopes.
+ */
+ public boolean hasEnvelopes()
+ {
+ return !envelopes.isEmpty();
+ }
+
+ /**
+ * Removes one envelope from the received messages queue or
+ * blocks until one is available.
+ */
+ public Envelope read() throws InterruptedException
+ {
+ return envelopes.take();
+ }
+
+ /**
+ * Removes and returnsn one endpoint event from the event queue or
+ * null if there are no endpoint events.
+ */
+ public EndpointEvent nextEvent()
+ {
+ return endpointEvents.poll();
+ }
+
+ protected void addEvent( EndpointEvent e )
+ {
+ endpointEvents.add( e );
+ }
+
+ protected void addEnvelope( Envelope env )
+ {
+ if( !envelopes.offer( env ) ) {
+ throw new KernelException( "Critical error, could not enqueue envelope." );
+ }
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel;
+
+import java.nio.ByteBuffer;
+
+/**
+ * A single channel remote connection allowing the sending
+ * and receiving of data. As opposed to the Kernel, this will
+ * only ever receive data from one Endpoint and so bypasses
+ * the envelope wrapping.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public interface Connector
+{
+ /**
+ * Returns true if this connector is currently connected.
+ */
+ public boolean isConnected();
+
+ /**
+ * Closes the connection. Any subsequent attempts to read
+ * or write will fail with an exception.
+ */
+ public void close();
+
+ /**
+ * Returns true if there is currently data available for
+ * reading. Some connector implementations may not be able
+ * to answer this question accurately and will always return
+ * false.
+ */
+ public boolean available();
+
+ /**
+ * Reads a chunk of data from the connection, blocking if
+ * there is no data available. The buffer may only be valid
+ * until the next read() call is made. Callers should copy
+ * the data if they need it for longer than that.
+ *
+ * @return The data read or null if there is no more data
+ * because the connection is closed.
+ */
+ public ByteBuffer read();
+
+ /**
+ * Writes a chunk of data to the connection.
+ */
+ public void write( ByteBuffer data );
+}
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel;
+
+
+/**
+ * Represents a client-side connection error, usually encapsulating
+ * an IOException as its cause.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public class ConnectorException extends RuntimeException
+{
+ public ConnectorException( String message, Throwable cause )
+ {
+ super( message, cause );
+ }
+
+ public ConnectorException( String message )
+ {
+ super( message );
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel;
+
+import java.nio.ByteBuffer;
+
+/**
+ * An abstract endpoint in a Kernel that can be used for
+ * sending/receiving messages within the kernel space.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public interface Endpoint
+{
+ /**
+ * Returns an ID that is unique for this endpoint within its
+ * Kernel instance.
+ */
+ public long getId();
+
+ /**
+ * Returns the kernel to which this endpoint belongs.
+ */
+ public Kernel getKernel();
+
+ /**
+ * Returns true if this endpoint is currently connected.
+ */
+ public boolean isConnected();
+
+ /**
+ * Sends data to the other end of the connection represented
+ * by this endpoint.
+ */
+ public void send( ByteBuffer data );
+
+ /**
+ * Closes this endpoint.
+ */
+ public void close();
+}
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel;
+
+
+/**
+ * Provides information about an added or
+ * removed connection.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public class EndpointEvent
+{
+ public enum Type { ADD, REMOVE };
+
+ private Kernel source;
+ private Endpoint endpoint;
+ private Type type;
+
+ public EndpointEvent( Kernel source, Endpoint p, Type type )
+ {
+ this.source = source;
+ this.endpoint = p;
+ this.type = type;
+ }
+
+ public static EndpointEvent createAdd( Kernel source, Endpoint p )
+ {
+ return new EndpointEvent( source, p, Type.ADD );
+ }
+
+ public static EndpointEvent createRemove( Kernel source, Endpoint p )
+ {
+ return new EndpointEvent( source, p, Type.REMOVE );
+ }
+
+ public Kernel getSource()
+ {
+ return source;
+ }
+
+ public Endpoint getEndpoint()
+ {
+ return endpoint;
+ }
+
+ public Type getType()
+ {
+ return type;
+ }
+
+ public String toString()
+ {
+ return "EndpointEvent[" + type + ", " + endpoint + "]";
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Encapsulates a received piece of data. This is used by the Kernel
+ * to track incoming chunks of data.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public class Envelope
+{
+ private Endpoint source;
+ private byte[] data;
+ private boolean reliable;
+
+ /**
+ * Creates an incoming envelope holding the data from the specified
+ * source. The 'reliable' flag further indicates on which mode of
+ * transport the data arrrived.
+ */
+ public Envelope( Endpoint source, byte[] data, boolean reliable )
+ {
+ this.source = source;
+ this.data = data;
+ this.reliable = reliable;
+ }
+
+ public Endpoint getSource()
+ {
+ return source;
+ }
+
+ public byte[] getData()
+ {
+ return data;
+ }
+
+ public boolean isReliable()
+ {
+ return reliable;
+ }
+
+ public String toString()
+ {
+ return "Envelope[" + source + ", " + (reliable?"reliable":"unreliable") + ", " + data.length + "]";
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+
+/**
+ * Defines the basic byte[] passing messaging
+ * kernel.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public interface Kernel
+{
+ /**
+ * Initializes the kernel and starts any internal processing.
+ */
+ public void initialize();
+
+ /**
+ * Gracefully terminates the kernel and stops any internal
+ * daemon processing. This method will not return until all
+ * internal threads have been shut down.
+ */
+ public void terminate() throws InterruptedException;
+
+ /**
+ * Dispatches the data to all endpoints managed by the
+ * kernel. 'routing' is currently ignored.
+ */
+ public void broadcast( Object routing, ByteBuffer data, boolean reliable );
+
+ /**
+ * Returns true if there are waiting envelopes.
+ */
+ public boolean hasEnvelopes();
+
+ /**
+ * Removes one envelope from the received messages queue or
+ * blocks until one is available.
+ */
+ public Envelope read() throws InterruptedException;
+
+ /**
+ * Removes and returnsn one endpoint event from the event queue or
+ * null if there are no endpoint events.
+ */
+ public EndpointEvent nextEvent();
+}
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel;
+
+
+/**
+ * Represents a kernel-level error, usually encapsulating
+ * an IOException as its cause.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public class KernelException extends RuntimeException
+{
+ public KernelException( String message, Throwable cause )
+ {
+ super( message, cause );
+ }
+
+ public KernelException( String message )
+ {
+ super( message );
+ }
+}
--- /dev/null
+<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">
+<html>
+<body>
+The kernel package is the heart of the JME networking module
+and controls the routing and dispatch of message data over
+different transport implementations. Most users will never
+have to deal with these classes unless they are writing their own
+client and server implementations that diverge from the standard
+classes that are provided.
+
+<p>{@link Kernel} defines the core of a server-side message
+broker that abstracts away the specific transport and underlying
+threading model used. For example, it might use NIO selectors
+in a single threaded model or straight multithreaded socket
+model. Or it might implement SSL connections. Once created,
+{@link Kernel} users don't need to care about the details.</p>
+
+<p>{@link Endpoint} is a managed connection within a {@link Kernel}
+providing kernel to client connectivity.</p>
+
+<p>{@link Connector} defines the basic client-side message sender
+and these objects are typically used to connect to a {@link Kernel} though
+they can connect to any network port that supports the implementation's
+protocol. Implementations are provided for straight TCP and UDP communication
+and could be extended to support SSL or different threading models.</p>
+
+</body>
+</html>
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel.tcp;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import com.jme3.network.kernel.*;
+
+
+/**
+ * Endpoint implementation that encapsulates the
+ * channel IO based connection information and keeps
+ * track of the outbound data queue for the channel.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public class NioEndpoint implements Endpoint
+{
+ private long id;
+ private SocketChannel socket;
+ private SelectorKernel kernel;
+ private ConcurrentLinkedQueue<ByteBuffer> outbound = new ConcurrentLinkedQueue<ByteBuffer>();
+
+ public NioEndpoint( SelectorKernel kernel, long id, SocketChannel socket )
+ {
+ this.id = id;
+ this.socket = socket;
+ this.kernel = kernel;
+ }
+
+ public Kernel getKernel()
+ {
+ return kernel;
+ }
+
+ public void close()
+ {
+ try {
+ kernel.closeEndpoint(this);
+ } catch( IOException e ) {
+ throw new KernelException( "Error closing endpoint for socket:" + socket, e );
+ }
+ }
+
+ public long getId()
+ {
+ return id;
+ }
+
+ public boolean isConnected()
+ {
+ return socket.isConnected();
+ }
+
+ /**
+ * The wakeup option is used internally when the kernel is
+ * broadcasting out to a bunch of endpoints and doesn't want to
+ * necessarily wakeup right away.
+ */
+ protected void send( ByteBuffer data, boolean copy, boolean wakeup )
+ {
+ // We create a ByteBuffer per endpoint since we
+ // use it to track the data sent to each endpoint
+ // separately.
+ ByteBuffer buffer;
+ if( !copy ) {
+ buffer = data;
+ } else {
+ // Copy the buffer
+ buffer = ByteBuffer.allocate(data.remaining());
+ buffer.put(data);
+ buffer.flip();
+ }
+
+ // Queue it up
+ outbound.add(buffer);
+
+ if( wakeup )
+ kernel.wakeupSelector();
+ }
+
+ /**
+ * Called by the SelectorKernel to get the current top
+ * buffer for writing.
+ */
+ protected ByteBuffer peekPending()
+ {
+ return outbound.peek();
+ }
+
+ /**
+ * Called by the SelectorKernel when the top buffer
+ * has been exhausted.
+ */
+ protected ByteBuffer removePending()
+ {
+ return outbound.poll();
+ }
+
+ protected boolean hasPending()
+ {
+ return !outbound.isEmpty();
+ }
+
+ public void send( ByteBuffer data )
+ {
+ send( data, true, true );
+ }
+
+ public String toString()
+ {
+ return "NioEndpoint[" + id + ", " + socket + "]";
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel.tcp;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.*;
+import java.nio.channels.spi.SelectorProvider;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.jme3.network.kernel.*;
+
+
+/**
+ * A Kernel implementation based on NIO selectors.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public class SelectorKernel extends AbstractKernel
+{
+ private InetAddress host;
+ private int port;
+ private SelectorThread thread;
+
+ private Map<Long,NioEndpoint> endpoints = new ConcurrentHashMap<Long,NioEndpoint>();
+
+ public SelectorKernel( InetAddress host, int port )
+ {
+ this.host = host;
+ this.port = port;
+ }
+
+ public SelectorKernel( int port ) throws IOException
+ {
+ this( InetAddress.getLocalHost(), port );
+ }
+
+ protected SelectorThread createSelectorThread()
+ {
+ return new SelectorThread();
+ }
+
+ public void initialize()
+ {
+ if( thread != null )
+ throw new IllegalStateException( "Kernel already initialized." );
+
+ thread = createSelectorThread();
+
+ try {
+ thread.connect();
+ thread.start();
+ } catch( IOException e ) {
+ throw new KernelException( "Error hosting:" + host + " port:" + port, e );
+ }
+ }
+
+ public void terminate() throws InterruptedException
+ {
+ if( thread == null )
+ throw new IllegalStateException( "Kernel not initialized." );
+
+ try {
+ thread.close();
+ thread = null;
+ } catch( IOException e ) {
+ throw new KernelException( "Error closing host connection:" + host + " port:" + port, e );
+ }
+ }
+
+ public void broadcast( Object routing, ByteBuffer data, boolean reliable )
+ {
+ if( !reliable )
+ throw new UnsupportedOperationException( "Unreliable send not supported by this kernel." );
+
+ // Copy the data just once
+ byte[] copy = new byte[data.remaining()];
+ System.arraycopy(data.array(), data.position(), copy, 0, data.remaining());
+
+ // Hand it to all of the endpoints that match our routing
+ for( NioEndpoint p : endpoints.values() ) {
+ // Does it match the routing information? FIXME
+
+ // Give it the data
+ p.send( data, false, false );
+ }
+
+ // Wake up the selector so it can reinitialize its
+ // state accordingly.
+ wakeupSelector();
+ }
+
+ protected NioEndpoint addEndpoint( SocketChannel c )
+ {
+ // Note: we purposely do NOT put the key in the endpoint.
+ // SelectionKeys are dangerous outside the selector thread
+ // and this is safer.
+ NioEndpoint p = new NioEndpoint( this, nextEndpointId(), c );
+
+ endpoints.put( p.getId(), p );
+
+ // Enqueue an endpoint event for the listeners
+ addEvent( EndpointEvent.createAdd( this, p ) );
+
+ return p;
+ }
+
+ protected void removeEndpoint( NioEndpoint p, SocketChannel c )
+ {
+ endpoints.remove( p.getId() );
+
+ // Enqueue an endpoint event for the listeners
+ addEvent( EndpointEvent.createRemove( this, p ) );
+ }
+
+ /**
+ * Called by the endpoints when they need to be closed.
+ */
+ protected void closeEndpoint( NioEndpoint p ) throws IOException
+ {
+ thread.cancel(p);
+ }
+
+ /**
+ * Used internally by the endpoints to wakeup the selector
+ * when they have data to send.
+ */
+ protected void wakeupSelector()
+ {
+ thread.wakeupSelector();
+ }
+
+ protected void newData( NioEndpoint p, SocketChannel c, ByteBuffer shared, int size )
+ {
+ // Note: if ever desirable, it would be possible to accumulate
+ // data per source channel and only 'finalize' it when
+ // asked for more envelopes then were ready. I just don't
+ // think it will be an issue in practice. The busier the
+ // server, the more the buffers will fill before we get to them.
+ // And if the server isn't busy, who cares if we chop things up
+ // smaller... the network is still likely to deliver things in
+ // bulk anyway.
+
+ // Must copy the shared data before we use it
+ byte[] dataCopy = new byte[size];
+ System.arraycopy(shared.array(), 0, dataCopy, 0, size);
+
+ Envelope env = new Envelope( p, dataCopy, true );
+ addEnvelope( env );
+ }
+
+ /**
+ * This class is purposely tucked neatly away because
+ * messing with the selector from other threads for any
+ * reason is very bad. This is the safest architecture.
+ */
+ protected class SelectorThread extends Thread
+ {
+ private ServerSocketChannel serverChannel;
+ private Selector selector;
+ private AtomicBoolean go = new AtomicBoolean(true);
+ private ByteBuffer working = ByteBuffer.allocate( 8192 );
+
+ /**
+ * Because we want to keep the keys to ourselves, we'll do
+ * the endpoint -> key mapping internally.
+ */
+ private Map<NioEndpoint,SelectionKey> endpointKeys = new ConcurrentHashMap<NioEndpoint,SelectionKey>();
+
+ public SelectorThread()
+ {
+ setName( "Selector@" + host + ":" + port );
+ setDaemon(true);
+ }
+
+ public void connect() throws IOException
+ {
+ // Create a new selector
+ this.selector = SelectorProvider.provider().openSelector();
+
+ // Create a new non-blocking server socket channel
+ this.serverChannel = ServerSocketChannel.open();
+ serverChannel.configureBlocking(false);
+
+ // Bind the server socket to the specified address and port
+ InetSocketAddress isa = new InetSocketAddress(host, port);
+ serverChannel.socket().bind(isa);
+
+ // Register the server socket channel, indicating an interest in
+ // accepting new connections
+ serverChannel.register(selector, SelectionKey.OP_ACCEPT);
+ }
+
+ public void close() throws IOException, InterruptedException
+ {
+ // Set the thread to stop
+ go.set(false);
+
+ // Make sure the channel is closed
+ serverChannel.close();
+
+ // And wait for it
+ join();
+ }
+
+ protected void wakeupSelector()
+ {
+ selector.wakeup();
+ }
+
+ protected void setupSelectorOptions()
+ {
+ // For now, selection keys will either be in OP_READ
+ // or OP_WRITE. So while we are writing a buffer, we
+ // will not be reading. This is way simpler and less
+ // error prone... it can always be changed when everything
+ // else works if we are looking to micro-optimize.
+
+ // Setup options based on the current state of
+ // the endpoints. This could potentially be more
+ // efficiently done as change requests... or simply
+ // keeping a thread-safe set of endpoints with pending
+ // writes. For most cases, it shouldn't matter.
+ for( Map.Entry<NioEndpoint,SelectionKey> e : endpointKeys.entrySet() ) {
+ if( e.getKey().hasPending() )
+ e.getValue().interestOps(SelectionKey.OP_WRITE);
+ }
+ }
+
+ protected void accept( SelectionKey key ) throws IOException
+ {
+ // Would only get accepts on a server channel
+ ServerSocketChannel serverChan = (ServerSocketChannel)key.channel();
+
+ // Setup the connection to be non-blocking
+ SocketChannel remoteChan = serverChan.accept();
+ remoteChan.configureBlocking(false);
+
+ // And disable Nagle's buffering algorithm... we want
+ // data to go when we put it there.
+ Socket sock = remoteChan.socket();
+ sock.setTcpNoDelay(true);
+
+ // Let the selector know we're interested in reading
+ // data from the channel
+ SelectionKey endKey = remoteChan.register( selector, SelectionKey.OP_READ );
+
+ // And now create a new endpoint
+ NioEndpoint p = addEndpoint( remoteChan );
+ endKey.attach(p);
+ endpointKeys.put(p, endKey);
+ }
+
+ protected void cancel( NioEndpoint p ) throws IOException
+ {
+ SelectionKey key = endpointKeys.remove(p);
+ SocketChannel c = (SocketChannel)key.channel();
+
+ // Note: key.cancel() is specifically thread safe. One of
+ // the few things one can do with a key from another
+ // thread.
+ key.cancel();
+ c.close();
+ removeEndpoint( p, c );
+ }
+
+ protected void cancel( SelectionKey key, SocketChannel c ) throws IOException
+ {
+ NioEndpoint p = (NioEndpoint)key.attachment();
+ endpointKeys.remove(p);
+
+ key.cancel();
+ c.close();
+ removeEndpoint( p, c );
+ }
+
+ protected void read( SelectionKey key ) throws IOException
+ {
+ NioEndpoint p = (NioEndpoint)key.attachment();
+ SocketChannel c = (SocketChannel)key.channel();
+ working.clear();
+
+ int size;
+ try {
+ size = c.read(working);
+ } catch( IOException e ) {
+ // The remove end forcibly closed the connection...
+ // close out our end and cancel the key
+ cancel( key, c );
+ return;
+ }
+
+ if( size == -1 ) {
+ // The remote end shut down cleanly...
+ // close out our end and cancel the key
+ cancel( key, c );
+ return;
+ }
+
+ newData( p, c, working, size );
+ }
+
+ protected void write( SelectionKey key ) throws IOException
+ {
+ NioEndpoint p = (NioEndpoint)key.attachment();
+ SocketChannel c = (SocketChannel)key.channel();
+
+ // We will send what we can and move on.
+ ByteBuffer current = p.peekPending();
+ c.write( current );
+
+ // If we wrote all of that packet then we need to remove it
+ if( current.remaining() == 0 ) {
+ p.removePending();
+ }
+
+ // If we happened to empty the pending queue then let's read
+ // again.
+ if( !p.hasPending() ) {
+ key.interestOps( SelectionKey.OP_READ );
+ }
+ }
+
+ protected void select() throws IOException
+ {
+ selector.select();
+
+ for( Iterator<SelectionKey> i = selector.selectedKeys().iterator(); i.hasNext(); ) {
+ SelectionKey key = i.next();
+ i.remove();
+
+ if( !key.isValid() )
+ continue;
+
+ if( key.isAcceptable() )
+ accept(key);
+ else if( key.isWritable() )
+ write(key);
+ else if( key.isReadable() )
+ read(key);
+ }
+ }
+
+ public void run()
+ {
+ // An atomic is safest and costs almost nothing
+ while( go.get() ) {
+ // Setup any queued option changes
+ setupSelectorOptions();
+
+ // Check for available keys and process them
+ try {
+ select();
+ } catch( ClosedSelectorException e ) {
+ if( !go.get() )
+ return; // it's because we're shutting down
+ throw new KernelException( "Premature selector closing", e );
+ } catch( IOException e ) {
+ if( !go.get() )
+ return; // error likely due to shutting down
+ reportError( e );
+ }
+ }
+ }
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel.tcp;
+
+import java.io.*;
+import java.net.*;
+import java.nio.ByteBuffer;
+
+import com.jme3.network.kernel.*;
+
+
+/**
+ * A straight forward socket-based connector implementation that
+ * does not use any separate threading. It relies completely on
+ * the buffering in the OS network layer.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public class SocketConnector implements Connector
+{
+ private Socket sock;
+ private InputStream in;
+ private OutputStream out;
+ private SocketAddress remoteAddress;
+ private byte[] buffer = new byte[65535];
+
+ public SocketConnector( InetAddress address, int port ) throws IOException
+ {
+ this.sock = new Socket(address, port);
+ remoteAddress = sock.getRemoteSocketAddress(); // for info purposes
+
+ // Disable Nagle's buffering so data goes out when we
+ // put it there.
+ sock.setTcpNoDelay(true);
+
+ in = sock.getInputStream();
+ out = sock.getOutputStream();
+ }
+
+ protected void checkClosed()
+ {
+ if( sock == null )
+ throw new ConnectorException( "Connection is closed:" + remoteAddress );
+ }
+
+ public boolean isConnected()
+ {
+ if( sock == null )
+ return false;
+ return sock.isConnected();
+ }
+
+ public void close()
+ {
+ checkClosed();
+ try {
+ Socket temp = sock;
+ sock = null;
+ temp.close();
+ } catch( IOException e ) {
+ throw new ConnectorException( "Error closing socket for:" + remoteAddress, e );
+ }
+ }
+
+ public boolean available()
+ {
+ checkClosed();
+ try {
+ return in.available() > 0;
+ } catch( IOException e ) {
+ throw new ConnectorException( "Error retrieving data availability for:" + remoteAddress, e );
+ }
+ }
+
+ public ByteBuffer read()
+ {
+ checkClosed();
+
+ try {
+ // Read what we can
+ int count = in.read(buffer);
+ if( count < 0 ) {
+ // Socket it closed
+ close();
+ return null;
+ }
+
+ // Wrap it in a ByteBuffer for the caller
+ return ByteBuffer.wrap( buffer, 0, count );
+ } catch( IOException e ) {
+ throw new ConnectorException( "Error reading from connection to:" + remoteAddress, e );
+ }
+ }
+
+ public void write( ByteBuffer data )
+ {
+ checkClosed();
+
+ try {
+ out.write(data.array(), data.position(), data.remaining());
+ } catch( IOException e ) {
+ throw new ConnectorException( "Error writing to connection:" + remoteAddress, e );
+ }
+ }
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel.udp;
+
+import java.io.*;
+import java.net.*;
+import java.nio.ByteBuffer;
+
+import com.jme3.network.kernel.*;
+
+
+/**
+ * A straight forward datagram socket-based UDP connector
+ * implementation.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public class UdpConnector implements Connector
+{
+ private DatagramSocket sock = new DatagramSocket();
+ private SocketAddress remoteAddress;
+ private byte[] buffer = new byte[65535];
+
+ /**
+ * In order to provide proper available() checking, we
+ * potentially queue one datagram.
+ */
+ private DatagramPacket pending;
+
+ /**
+ * Creates a new UDP connection that send datagrams to the
+ * specified address and port.
+ */
+ public UdpConnector( InetAddress local, int localPort,
+ InetAddress remote, int remotePort ) throws IOException
+ {
+ this.sock = new DatagramSocket( localPort, local );
+ remoteAddress = new InetSocketAddress( remote, remotePort );
+
+ // Setup to receive only from the remote address
+ sock.connect( remoteAddress );
+ }
+
+ protected void checkClosed()
+ {
+ if( sock == null )
+ throw new ConnectorException( "Connection is closed:" + remoteAddress );
+ }
+
+ public boolean isConnected()
+ {
+ if( sock == null )
+ return false;
+ return sock.isConnected();
+ }
+
+ public void close()
+ {
+ checkClosed();
+ DatagramSocket temp = sock;
+ sock = null;
+ temp.close();
+ }
+
+ /**
+ * This always returns false since the simple DatagramSocket usage
+ * cannot be run in a non-blocking way.
+ */
+ public boolean available()
+ {
+ // It would take a separate thread or an NIO Selector based implementation to get this
+ // to work. If a polling strategy is never employed by callers then it doesn't
+ // seem worth it to implement all of that just for this method.
+ checkClosed();
+ return false;
+ }
+
+ public ByteBuffer read()
+ {
+ checkClosed();
+
+ try {
+ DatagramPacket packet = new DatagramPacket( buffer, buffer.length );
+ sock.receive(packet);
+
+ // Wrap it in a ByteBuffer for the caller
+ return ByteBuffer.wrap( buffer, 0, packet.getLength() );
+ } catch( IOException e ) {
+ throw new ConnectorException( "Error reading from connection to:" + remoteAddress, e );
+ }
+ }
+
+ public void write( ByteBuffer data )
+ {
+ checkClosed();
+
+ try {
+ DatagramPacket p = new DatagramPacket( data.array(), data.position(), data.remaining(),
+ remoteAddress );
+ sock.send(p);
+ } catch( IOException e ) {
+ throw new ConnectorException( "Error writing to connection:" + remoteAddress, e );
+ }
+ }
+}
+
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel.udp;
+
+import java.io.IOException;
+import java.net.*;
+import java.nio.ByteBuffer;
+
+import com.jme3.network.kernel.*;
+
+
+/**
+ * Endpoint implementation that encapsulates the
+ * UDP connection information for return messaging,
+ * identification of envelope sources, etc.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public class UdpEndpoint implements Endpoint
+{
+ private long id;
+ private SocketAddress address;
+ private DatagramSocket socket;
+ private UdpKernel kernel;
+
+ public UdpEndpoint( UdpKernel kernel, long id, SocketAddress address, DatagramSocket socket )
+ {
+ this.id = id;
+ this.address = address;
+ this.socket = socket;
+ this.kernel = kernel;
+ }
+
+ public Kernel getKernel()
+ {
+ return kernel;
+ }
+
+ protected SocketAddress getRemoteAddress()
+ {
+ return address;
+ }
+
+ public void close()
+ {
+ try {
+ kernel.closeEndpoint(this);
+ } catch( IOException e ) {
+ throw new KernelException( "Error closing endpoint for socket:" + socket, e );
+ }
+ }
+
+ public long getId()
+ {
+ return id;
+ }
+
+ public boolean isConnected()
+ {
+ return socket.isConnected();
+ }
+
+ public void send( ByteBuffer data )
+ {
+ try {
+ DatagramPacket p = new DatagramPacket( data.array(), data.position(),
+ data.remaining(), address );
+ socket.send(p);
+ } catch( IOException e ) {
+ throw new KernelException( "Error sending datagram to:" + address, e );
+ }
+ }
+
+ public String toString()
+ {
+ return "UdpEndpoint[" + id + ", " + socket + "]";
+ }
+}
--- /dev/null
+/*
+ * Copyright (c) 2011 jMonkeyEngine
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * * Neither the name of 'jMonkeyEngine' nor the names of its contributors
+ * may be used to endorse or promote products derived from this software
+ * without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+ * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+ * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+ * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package com.jme3.network.kernel.udp;
+
+import java.io.IOException;
+import java.net.*;
+import java.nio.ByteBuffer;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.jme3.network.kernel.*;
+
+/**
+ * A Kernel implementation using UDP packets.
+ *
+ * @version $Revision$
+ * @author Paul Speed
+ */
+public class UdpKernel extends AbstractKernel
+{
+ private InetAddress host;
+ private int port;
+ private HostThread thread;
+
+ // The nature of UDP means that even through a firewall,
+ // a user would have to have a unique address+port since UDP
+ // can't really be NAT'ed.
+ private Map<SocketAddress,UdpEndpoint> socketEndpoints = new ConcurrentHashMap<SocketAddress,UdpEndpoint>();
+
+ public UdpKernel( InetAddress host, int port )
+ {
+ this.host = host;
+ this.port = port;
+ }
+
+ public UdpKernel( int port ) throws IOException
+ {
+ this( InetAddress.getLocalHost(), port );
+ }
+
+ protected HostThread createHostThread()
+ {
+ return new HostThread();
+ }
+
+ public void initialize()
+ {
+ if( thread != null )
+ throw new IllegalStateException( "Kernel already initialized." );
+
+ thread = createHostThread();
+
+ try {
+ thread.connect();
+ thread.start();
+ } catch( IOException e ) {
+ throw new KernelException( "Error hosting:" + host + " port:" + port, e );
+ }
+ }
+
+ public void terminate() throws InterruptedException
+ {
+ if( thread == null )
+ throw new IllegalStateException( "Kernel not initialized." );
+
+ try {
+ thread.close();
+ thread = null;
+ } catch( IOException e ) {
+ throw new KernelException( "Error closing host connection:" + host + " port:" + port, e );
+ }
+ }
+
+ /**
+ * Dispatches the data to all endpoints managed by the
+ * kernel. 'routing' is currently ignored.
+ */
+ public void broadcast( Object routing, ByteBuffer data, boolean reliable )
+ {
+ if( reliable )
+ throw new UnsupportedOperationException( "Reliable send not supported by this kernel." );
+
+ // Hand it to all of the endpoints that match our routing
+ for( UdpEndpoint p : socketEndpoints.values() ) {
+ // Does it match the routing information? FIXME
+
+ // Send the data
+ p.send( data );
+ }
+ }
+
+ protected Endpoint getEndpoint( SocketAddress address, boolean create )
+ {
+ UdpEndpoint p = socketEndpoints.get(address);
+ if( p == null && create ) {
+ p = new UdpEndpoint( this, nextEndpointId(), address, thread.getSocket() );
+ socketEndpoints.put( address, p );
+
+ // Add an event for it.
+ addEvent( EndpointEvent.createAdd( this, p ) );
+ }
+ return p;
+ }
+
+ /**
+ * Called by the endpoints when they need to be closed.
+ */
+ protected void closeEndpoint( UdpEndpoint p ) throws IOException
+ {
+ // Just book-keeping to do here.
+ socketEndpoints.remove( p.getRemoteAddress() );
+
+ addEvent( EndpointEvent.createRemove( this, p ) );
+ }
+
+ protected void newData( DatagramPacket packet )
+ {
+ // So the tricky part here is figuring out the endpoint and
+ // whether it's new or not. In these UDP schemes, firewalls have
+ // to be ported back to a specific machine so we will consider
+ // the address + port (ie: SocketAddress) the defacto unique
+ // ID.
+ Endpoint p = getEndpoint( packet.getSocketAddress(), true );
+
+ // We'll copy the data to trim it.
+ byte[] data = new byte[packet.getLength()];
+ System.arraycopy(packet.getData(), 0, data, 0, data.length);
+
+ Envelope env = new Envelope( p, data, false );
+ addEnvelope( env );
+ }
+
+ protected class HostThread extends Thread
+ {
+ private DatagramSocket socket;
+ private AtomicBoolean go = new AtomicBoolean(true);
+
+ private byte[] buffer = new byte[65535]; // slightly bigger than needed.
+
+ public HostThread()
+ {
+ setName( "UDP Host@" + host + ":" + port );
+ setDaemon(true);
+ }
+
+ protected DatagramSocket getSocket()
+ {
+ return socket;
+ }
+
+ public void connect() throws IOException
+ {
+ socket = new DatagramSocket( port, host );
+ }
+
+ public void close() throws IOException, InterruptedException
+ {
+ // Set the thread to stop
+ go.set(false);
+
+ // Make sure the channel is closed
+ socket.close();
+
+ // And wait for it
+ join();
+ }
+
+ public void run()
+ {
+ // An atomic is safest and costs almost nothing
+ while( go.get() ) {
+ try {
+ // Could reuse the packet but I don't see the
+ // point and it may lead to subtle bugs if not properly
+ // reset.
+ DatagramPacket packet = new DatagramPacket( buffer, buffer.length );
+ socket.receive(packet);
+
+ newData( packet );
+ } catch( IOException e ) {
+ if( !go.get() )
+ return;
+ reportError( e );
+ }
+ }
+ }
+ }
+}
OSDN Git Service
