[direct-certificate-discovery-tool] push by michal.kotelba@esacinc.com - - Supports DCDT-226.... on 2014-05-22 14:26 GMT
1 view
Skip to first unread message
direct-certifica...@googlecode.com
May 22, 2014, 10:26:07 AM5/22/14
to dcdt...@googlegroups.com
Revision: d0a260ccd4e4
Branch: default
Author: Michal Kotelba <michal....@esacinc.com>
Date: Thu May 22 06:39:17 2014 UTC
Log: - Supports DCDT-226.
- Refactored DNS service + underlying (N)IO framework to utilize adapted
sockets instead of NIO channels.
- Added DNS service concurrent load functional testing.
http://code.google.com/p/direct-certificate-discovery-tool/source/detail?r=d0a260ccd4e4
Added:
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/ClientSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketAdapterOptions.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketRequest.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketRequestProcessor.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpServerSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpSocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/ToolSocketOption.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/UdpSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/UdpSocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractClientSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketRequest.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketRequestProcessor.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractTcpSocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractUdpSocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/TcpServerSocketAdapterImpl.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/TcpSocketAdapterImpl.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/UdpSocketAdapterImpl.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/utils/ToolSocketUtils.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequest.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequestProcessingException.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequestProcessor.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerTcpSocketListener.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerUdpSocketListener.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerRequestImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerRequestProcessorImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerTcpSocketListenerImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerUdpSocketListenerImpl.java
/dcdt-service-dns/src/test/resources/META-INF/service/dns/service-dns-test.properties
Deleted:
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/ToolSocketOption.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerDataProcessor.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerDataProcessorCallback.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerSelector.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelSocketOptions.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/SelectionAttachment.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/SelectionOperationType.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerDataProcessor.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerDataProcessorCallback.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerSelector.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractSelectionAttachment.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractTcpChannelListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractUdpChannelListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/utils/ToolChannelUtils.java
/dcdt-core/src/main/resources/META-INF/spring/spring-core-nio.xml
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerChannelListenerSelector.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerQueryAttachment.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerQueryResolutionException.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerQueryResolver.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerQueryResolverCallback.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerChannelListenerSelectorImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerQueryAttachmentImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerQueryResolverCallbackImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerQueryResolverImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerTcpChannelListener.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerUdpChannelListener.java
Modified:
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/ToolNetException.java
/dcdt-core/src/main/resources/META-INF/spring/spring-core-net.xml
/dcdt-core/src/test/java/gov/hhs/onc/dcdt/config/instance/InstanceConfigRegistryFunctionalTests.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerImpl.java
/dcdt-service-dns/src/main/resources/META-INF/spring/spring-service-dns.xml
/dcdt-service-dns/src/test/java/gov/hhs/onc/dcdt/service/dns/DnsServiceFunctionalTests.java
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/ClientSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,22 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.SocketAddress;
+import javax.annotation.Nonnegative;
+import org.apache.commons.lang3.tuple.Pair;
+
+public interface ClientSocketAdapter<T extends Closeable> extends
SocketAdapter<T> {
+ public void write(byte[] data, SocketAddress remoteAddr) throws
IOException;
+
+ public Pair<SocketAddress, byte[]> read() throws IOException;
+
+ public Pair<SocketAddress, byte[]> read(@Nonnegative int bufferLen)
throws IOException;
+
+ public Pair<SocketAddress, byte[]> read(byte[] buffer) throws
IOException;
+
+ @Nonnegative
+ public int getSendBufferSize() throws IOException;
+
+ public void setSendBufferSize(@Nonnegative int sendBufferSize) throws
IOException;
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,40 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import gov.hhs.onc.dcdt.beans.ToolBean;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.net.SocketOption;
+import java.util.Map;
+import javax.annotation.Nonnegative;
+
+public interface SocketAdapter<T extends Closeable> extends Closeable,
ToolBean {
+ public void bind(SocketAddress bindSocketAddr) throws IOException;
+
+ public void setOptions(Map<? extends SocketOption<?>, ?> optMap)
throws IOException;
+
+ public <U> void setOption(SocketOption<U> optKey, U optValue) throws
IOException;
+
+ public boolean isBound();
+
+ public boolean isClosed();
+
+ public InetProtocol getProtocol();
+
+ @Nonnegative
+ public int getReceiveBufferSize() throws IOException;
+
+ public void setReceiveBufferSize(@Nonnegative int receiveBufferSize)
throws IOException;
+
+ public boolean getReuseAddress() throws IOException;
+
+ public void setReuseAddress(boolean reuseAddr) throws IOException;
+
+ public T getSocket();
+
+ @Nonnegative
+ public int getTimeout() throws IOException;
+
+ public void setTimeout(@Nonnegative int timeout) throws IOException;
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketAdapterOptions.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,10 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.net.SocketOption;
+
+public final class SocketAdapterOptions {
+ public final static SocketOption<Integer> TIMEOUT = new
ToolSocketOption<>("SO_TIMEOUT", Integer.class);
+
+ private SocketAdapterOptions() {
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,9 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import gov.hhs.onc.dcdt.beans.ToolLifecycleBean;
+import java.io.Closeable;
+import org.springframework.context.ApplicationContextAware;
+
+public interface SocketListener<T extends Closeable, U extends
SocketAdapter<T>, V extends Closeable, W extends ClientSocketAdapter<V>, X
extends SocketRequest, Y extends SocketRequestProcessor<X>>
+ extends ApplicationContextAware, ToolLifecycleBean {
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketRequest.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,20 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import gov.hhs.onc.dcdt.beans.ToolBean;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import javax.annotation.Nullable;
+
+public interface SocketRequest extends ToolBean {
+ public InetProtocol getProtocol();
+
+ @Nullable
+ public SocketAddress getRemoteAddress();
+
+ public void setRemoteAddress(@Nullable SocketAddress remoteAddr);
+
+ public ByteBuffer getRequestBuffer();
+
+ public ByteBuffer getResponseBuffer();
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketRequestProcessor.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,7 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import gov.hhs.onc.dcdt.beans.ToolBean;
+
+public interface SocketRequestProcessor<T extends SocketRequest> extends
ToolBean {
+ public byte[] processRequest();
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpServerSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,9 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+
+public interface TcpServerSocketAdapter extends
SocketAdapter<ServerSocket> {
+ public Socket accept() throws IOException;
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,8 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.io.IOException;
+import java.net.Socket;
+
+public interface TcpSocketAdapter extends ClientSocketAdapter<Socket> {
+ public void write(byte[] data) throws IOException;
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,8 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+
+public interface TcpSocketListener<T extends SocketRequest, U extends
SocketRequestProcessor<T>> extends
+ SocketListener<ServerSocket, TcpServerSocketAdapter, Socket,
TcpSocketAdapter, T, U> {
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/ToolSocketOption.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,23 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.net.SocketOption;
+
+public class ToolSocketOption<T> implements SocketOption<T> {
+ private String name;
+ private Class<T> type;
+
+ public ToolSocketOption(String name, Class<T> type) {
+ this.name = name;
+ this.type = type;
+ }
+
+ @Override
+ public String name() {
+ return this.name;
+ }
+
+ @Override
+ public Class<T> type() {
+ return this.type;
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/UdpSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,6 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.net.DatagramSocket;
+
+public interface UdpSocketAdapter extends
ClientSocketAdapter<DatagramSocket> {
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/UdpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,7 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.net.DatagramSocket;
+
+public interface UdpSocketListener<T extends SocketRequest, U extends
SocketRequestProcessor<T>> extends
+ SocketListener<DatagramSocket, UdpSocketAdapter, DatagramSocket,
UdpSocketAdapter, T, U> {
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractClientSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,22 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.sockets.ClientSocketAdapter;
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.SocketOption;
+import java.net.StandardSocketOptions;
+
+public abstract class AbstractClientSocketAdapter<T extends Closeable>
extends AbstractSocketAdapter<T> implements ClientSocketAdapter<T> {
+ protected AbstractClientSocketAdapter(T socket) {
+ super(socket);
+ }
+
+ @Override
+ protected <U> void setOptionInternal(SocketOption<U> optKey, U
optValue) throws IOException {
+ if (optKey == StandardSocketOptions.SO_SNDBUF) {
+ this.setSendBufferSize(((Integer) optValue));
+ } else {
+ super.setOptionInternal(optKey, optValue);
+ }
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,54 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.beans.impl.AbstractToolBean;
+import gov.hhs.onc.dcdt.net.sockets.SocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.SocketAdapterOptions;
+import gov.hhs.onc.dcdt.net.ToolNetException;
+import gov.hhs.onc.dcdt.utils.ToolClassUtils;
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.SocketOption;
+import java.net.StandardSocketOptions;
+import java.util.Map;
+import java.util.Map.Entry;
+
+public abstract class AbstractSocketAdapter<T extends Closeable> extends
AbstractToolBean implements SocketAdapter<T> {
+ protected T socket;
+
+ protected AbstractSocketAdapter(T socket) {
+ this.socket = socket;
+ }
+
+ @Override
+ @SuppressWarnings({ "unchecked" })
+ public void setOptions(Map<? extends SocketOption<?>, ?> optMap)
throws IOException {
+ for (Entry<? extends SocketOption<?>, ?> optEntry :
optMap.entrySet()) {
+ this.setOption(((SocketOption<Object>) optEntry.getKey()),
optEntry.getValue());
+ }
+ }
+
+ @Override
+ public <U> void setOption(SocketOption<U> optKey, U optValue) throws
IOException {
+ try {
+ this.setOptionInternal(optKey, optValue);
+ } catch (ToolNetException e) {
+ throw e;
+ } catch (IOException e) {
+ throw new ToolNetException(String.format("Unable to set socket
(class=%s) adapter (class=%s) option (key=%s) value: %s",
+ ToolClassUtils.getName(this.socket),
ToolClassUtils.getName(this), optKey, optValue), e);
+ }
+ }
+
+ protected <U> void setOptionInternal(SocketOption<U> optKey, U
optValue) throws IOException {
+ if (optKey == StandardSocketOptions.SO_RCVBUF) {
+ this.setReceiveBufferSize(((Integer) optValue));
+ } else if (optKey == StandardSocketOptions.SO_REUSEADDR) {
+ this.setReuseAddress(((Boolean) optValue));
+ } else if (optKey == SocketAdapterOptions.TIMEOUT) {
+ this.setTimeout(((Integer) optValue));
+ } else {
+ throw new ToolNetException(String.format("Unable to set
unknown socket (class=%s) adapter (class=%s) option (key=%s) value: %s",
+ ToolClassUtils.getName(this.socket),
ToolClassUtils.getName(this), optKey, optValue));
+ }
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,181 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.beans.impl.AbstractToolLifecycleBean;
+import gov.hhs.onc.dcdt.beans.utils.ToolBeanFactoryUtils;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.ClientSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.SocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.SocketListener;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequestProcessor;
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.FutureTask;
+import javax.annotation.Nullable;
+import org.apache.mina.util.ConcurrentHashSet;
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.util.concurrent.ListenableFutureCallback;
+import org.springframework.util.concurrent.ListenableFutureTask;
+
+public abstract class AbstractSocketListener<T extends Closeable, U
extends SocketAdapter<T>, V extends Closeable, W extends
ClientSocketAdapter<V>, X extends SocketRequest, Y extends
SocketRequestProcessor<X>>
+ extends AbstractToolLifecycleBean implements SocketListener<T, U, V,
W, X, Y> {
+ protected class SocketListenDaemon implements Callable<Void> {
+ @Nullable
+ @Override
+ public Void call() throws Exception {
+ while ((AbstractSocketListener.this.listenSocketAdapter !=
null) && !AbstractSocketListener.this.listenSocketAdapter.isClosed()) {
+ X req =
AbstractSocketListener.this.createRequest(AbstractSocketListener.this.listenSocketAdapter.getProtocol());
+
+ final W reqSocketAdapter =
AbstractSocketListener.this.readRequest(AbstractSocketListener.this.listenSocketAdapter,
req);
+
AbstractSocketListener.this.reqSocketAdapters.add(reqSocketAdapter);
+
+ final ListenableFutureTask<Void> reqDaemonTask =
AbstractSocketListener.this.createRequestDaemonTask(reqSocketAdapter, req);
+
AbstractSocketListener.this.reqDaemonTasks.add(reqDaemonTask);
+
+ reqDaemonTask.addCallback(new
ListenableFutureCallback<Void>() {
+ @Override
+ public void onSuccess(@Nullable Void result) {
+ this.processResponse();
+ }
+
+ @Override
+ public void onFailure(Throwable th) {
+ this.processResponse();
+ }
+
+ private void processResponse() {
+
AbstractSocketListener.this.reqDaemonTasks.remove(reqDaemonTask);
+
AbstractSocketListener.this.reqSocketAdapters.remove(reqSocketAdapter);
+ }
+ });
+
+
AbstractSocketListener.this.taskExec.execute(reqDaemonTask);
+ }
+
+ return null;
+ }
+ }
+
+ protected class SocketRequestDaemon implements Callable<Void> {
+ protected W reqSocketAdapter;
+ protected X req;
+
+ protected SocketRequestDaemon(W reqSocketAdapter, X req) {
+ this.reqSocketAdapter = reqSocketAdapter;
+ this.req = req;
+ }
+
+ @Nullable
+ @Override
+ public Void call() throws Exception {
+
AbstractSocketListener.this.writeResponse(this.reqSocketAdapter,
AbstractSocketListener.this.createRequestProcessor(req).processRequest(),
+ req.getRemoteAddress());
+
+ return null;
+ }
+ }
+
+ protected Class<T> listenSocketClass;
+ protected Class<U> listenSocketAdapterClass;
+ protected Class<V> reqSocketClass;
+ protected Class<W> reqSocketAdapterClass;
+ protected Class<X> reqClass;
+ protected Class<Y> reqProcClass;
+ protected SocketAddress bindSocketAddr;
+ protected AbstractApplicationContext appContext;
+ protected U listenSocketAdapter;
+ protected Set<W> reqSocketAdapters = new ConcurrentHashSet<>();
+ protected ListenableFutureTask<Void> listenDaemonTask;
+ protected Set<ListenableFutureTask<Void>> reqDaemonTasks = new
ConcurrentHashSet<>();
+
+ protected AbstractSocketListener(Class<T> listenSocketClass, Class<U>
listenSocketAdapterClass, Class<V> reqSocketClass, Class<W>
reqSocketAdapterClass,
+ Class<X> reqClass, Class<Y> reqProcClass, SocketAddress
bindSocketAddr) {
+ this.listenSocketClass = listenSocketClass;
+ this.listenSocketAdapterClass = listenSocketAdapterClass;
+ this.reqSocketClass = reqSocketClass;
+ this.reqSocketAdapterClass = reqSocketAdapterClass;
+ this.reqClass = reqClass;
+ this.reqProcClass = reqProcClass;
+ this.bindSocketAddr = bindSocketAddr;
+ }
+
+ @Override
+ public boolean isRunning() {
+ return (super.isRunning() && (((this.listenSocketAdapter != null)
&& !listenSocketAdapter.isClosed()) || !this.reqSocketAdapters.isEmpty()
+ || ((this.listenDaemonTask != null)
&& !this.listenDaemonTask.isDone()) || !this.reqDaemonTasks.isEmpty()));
+ }
+
+ @Override
+ protected void stopInternal() throws Exception {
+ if ((this.listenDaemonTask != null)
&& !this.listenDaemonTask.isDone()) {
+ this.listenDaemonTask.cancel(true);
+ }
+
+ if ((this.listenSocketAdapter != null)
&& !this.listenSocketAdapter.isClosed()) {
+ this.listenSocketAdapter.close();
+ }
+
+ for (FutureTask<Void> reqDaemonTask : this.reqDaemonTasks) {
+ if (!reqDaemonTask.isDone()) {
+ reqDaemonTask.cancel(true);
+ }
+ }
+
+ for (W reqSocketAdapter : this.reqSocketAdapters) {
+ if (!reqSocketAdapter.isClosed()) {
+ reqSocketAdapter.close();
+ }
+ }
+ }
+
+ @Override
+ protected void startInternal() throws Exception {
+ (this.listenSocketAdapter =
this.createListenSocketAdapter(this.createListenSocket())).bind(this.bindSocketAddr);
+
+ this.taskExec.execute((this.listenDaemonTask =
this.createListenDaemonTask()));
+ }
+
+ protected void writeResponse(W reqSocketAdapter, byte[] respData,
SocketAddress remoteAddr) throws IOException {
+ reqSocketAdapter.write(respData, remoteAddr);
+ }
+
+ protected Y createRequestProcessor(Object ... reqProcArgs) {
+ return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.reqProcClass, reqProcArgs);
+ }
+
+ protected abstract W readRequest(U listenSocketAdapter, X req) throws
IOException;
+
+ protected X createRequest(InetProtocol protocol) {
+ return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.reqClass, protocol);
+ }
+
+ protected ListenableFutureTask<Void> createRequestDaemonTask(W
respSocketAdapter, X req) {
+ return new ListenableFutureTask<>(new
SocketRequestDaemon(respSocketAdapter, req));
+ }
+
+ protected ListenableFutureTask<Void> createListenDaemonTask() {
+ return new ListenableFutureTask<>(new SocketListenDaemon());
+ }
+
+ protected W createRequestSocketAdapter(V reqSocket) throws IOException
{
+ return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.reqSocketAdapterClass, reqSocket);
+ }
+
+ protected U createListenSocketAdapter(T listenSocket) throws
IOException {
+ return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.listenSocketAdapterClass, listenSocket);
+ }
+
+ protected T createListenSocket(Object ... listenSocketArgs) {
+ return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.listenSocketClass, listenSocketArgs);
+ }
+
+ @Override
+ public void setApplicationContext(ApplicationContext appContext)
throws BeansException {
+ this.appContext = ((AbstractApplicationContext) appContext);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketRequest.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,48 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.beans.impl.AbstractToolBean;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import javax.annotation.Nonnegative;
+import javax.annotation.Nullable;
+
+public abstract class AbstractSocketRequest extends AbstractToolBean
implements SocketRequest {
+ protected InetProtocol protocol;
+ protected SocketAddress remoteAddr;
+ protected ByteBuffer reqBuffer;
+ protected ByteBuffer respBuffer;
+
+ protected AbstractSocketRequest(InetProtocol protocol, @Nonnegative
int reqBufferSize, @Nonnegative int respBufferSize) {
+ this.protocol = protocol;
+ this.reqBuffer = ByteBuffer.allocate(reqBufferSize);
+ this.respBuffer = ByteBuffer.allocate(respBufferSize);
+ }
+
+ @Override
+ public InetProtocol getProtocol() {
+ return this.protocol;
+ }
+
+ @Nullable
+ @Override
+ public SocketAddress getRemoteAddress() {
+ return this.remoteAddr;
+ }
+
+ @Override
+ public void setRemoteAddress(@Nullable SocketAddress remoteAddr) {
+ this.remoteAddr = remoteAddr;
+ }
+
+ @Override
+ public ByteBuffer getRequestBuffer() {
+ return this.reqBuffer;
+ }
+
+ @Override
+ public ByteBuffer getResponseBuffer() {
+ return this.respBuffer;
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketRequestProcessor.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,33 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.beans.impl.AbstractToolBean;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequestProcessor;
+import gov.hhs.onc.dcdt.nio.utils.ToolBufferUtils;
+import java.nio.ByteBuffer;
+
+public abstract class AbstractSocketRequestProcessor<T extends
SocketRequest> extends AbstractToolBean implements
SocketRequestProcessor<T> {
+ protected T req;
+
+ protected AbstractSocketRequestProcessor(T req) {
+ this.req = req;
+ }
+
+ @Override
+ public byte[] processRequest() {
+ byte[] reqData =
ToolBufferUtils.get(ToolBufferUtils.flip(this.req.getRequestBuffer()));
+ ByteBuffer respBuffer = this.req.getResponseBuffer();
+
+ try {
+ respBuffer.put(this.processRequestInternal(reqData));
+ } catch (Exception e) {
+ respBuffer.put(this.processError(reqData, e));
+ }
+
+ return ToolBufferUtils.get(ToolBufferUtils.flip(respBuffer));
+ }
+
+ protected abstract byte[] processError(byte[] reqData, Exception
exception);
+
+ protected abstract byte[] processRequestInternal(byte[] reqData)
throws Exception;
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractTcpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,33 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequestProcessor;
+import gov.hhs.onc.dcdt.net.sockets.TcpServerSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.TcpSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.TcpSocketListener;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import org.apache.commons.lang3.tuple.Pair;
+
+public class AbstractTcpSocketListener<T extends SocketRequest, U extends
SocketRequestProcessor<T>> extends
+ AbstractSocketListener<ServerSocket, TcpServerSocketAdapter, Socket,
TcpSocketAdapter, T, U> implements TcpSocketListener<T, U> {
+ protected AbstractTcpSocketListener(Class<T> reqClass, Class<U>
reqProcClass, SocketAddress bindSocketAddr) {
+ super(ServerSocket.class, TcpServerSocketAdapter.class,
Socket.class, TcpSocketAdapter.class, reqClass, reqProcClass,
bindSocketAddr);
+ }
+
+ @Override
+ protected TcpSocketAdapter readRequest(TcpServerSocketAdapter
listenSocketAdapter, T req) throws IOException {
+ TcpSocketAdapter reqSocketAdapter =
this.createRequestSocketAdapter(listenSocketAdapter.accept());
+ ByteBuffer reqBuffer = req.getRequestBuffer();
+ Pair<SocketAddress, byte[]> reqPair =
reqSocketAdapter.read(reqBuffer.remaining());
+
+ req.setRemoteAddress(reqPair.getLeft());
+
+ reqBuffer.put(reqPair.getRight());
+
+ return reqSocketAdapter;
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractUdpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,35 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequestProcessor;
+import gov.hhs.onc.dcdt.net.sockets.UdpSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.UdpSocketListener;
+import java.io.IOException;
+import java.net.DatagramSocket;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import org.apache.commons.lang3.tuple.Pair;
+
+public abstract class AbstractUdpSocketListener<T extends SocketRequest, U
extends SocketRequestProcessor<T>> extends
+ AbstractSocketListener<DatagramSocket, UdpSocketAdapter,
DatagramSocket, UdpSocketAdapter, T, U> implements UdpSocketListener<T, U> {
+ protected AbstractUdpSocketListener(Class<T> reqClass, Class<U>
reqProcClass, SocketAddress bindSocketAddr) {
+ super(DatagramSocket.class, UdpSocketAdapter.class,
DatagramSocket.class, UdpSocketAdapter.class, reqClass, reqProcClass,
bindSocketAddr);
+ }
+
+ @Override
+ protected UdpSocketAdapter readRequest(UdpSocketAdapter
listenSocketAdapter, T req) throws IOException {
+ ByteBuffer reqBuffer = req.getRequestBuffer();
+ Pair<SocketAddress, byte[]> reqPair =
listenSocketAdapter.read(reqBuffer.remaining());
+
+ req.setRemoteAddress(reqPair.getLeft());
+
+ reqBuffer.put(reqPair.getRight());
+
+ return listenSocketAdapter;
+ }
+
+ @Override
+ protected DatagramSocket createListenSocket(Object ...
listenSocketArgs) {
+ return super.createListenSocket(((SocketAddress) null));
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/TcpServerSocketAdapterImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,80 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.TcpServerSocketAdapter;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketAddress;
+import javax.annotation.Nonnegative;
+
+public class TcpServerSocketAdapterImpl extends
AbstractSocketAdapter<ServerSocket> implements TcpServerSocketAdapter {
+ public TcpServerSocketAdapterImpl(ServerSocket socket) {
+ super(socket);
+ }
+
+ @Override
+ public Socket accept() throws IOException {
+ return this.socket.accept();
+ }
+
+ @Override
+ public void bind(SocketAddress bindSocketAddr) throws IOException {
+ this.socket.bind(bindSocketAddr);
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.socket.close();
+ }
+
+ @Override
+ public boolean isBound() {
+ return this.socket.isBound();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return this.socket.isClosed();
+ }
+
+ @Override
+ public InetProtocol getProtocol() {
+ return InetProtocol.TCP;
+ }
+
+ @Override
+ public int getReceiveBufferSize() throws IOException {
+ return this.socket.getReceiveBufferSize();
+ }
+
+ @Override
+ public void setReceiveBufferSize(@Nonnegative int receiveBufferSize)
throws IOException {
+ this.socket.setReceiveBufferSize(receiveBufferSize);
+ }
+
+ @Override
+ public boolean getReuseAddress() throws IOException {
+ return this.socket.getReuseAddress();
+ }
+
+ @Override
+ public void setReuseAddress(boolean reuseAddr) throws IOException {
+ this.socket.setReuseAddress(reuseAddr);
+ }
+
+ @Override
+ public ServerSocket getSocket() {
+ return this.socket;
+ }
+
+ @Override
+ public int getTimeout() throws IOException {
+ return this.socket.getSoTimeout();
+ }
+
+ @Override
+ public void setTimeout(@Nonnegative int timeout) throws IOException {
+ this.socket.setSoTimeout(timeout);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/TcpSocketAdapterImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,120 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.TcpSocketAdapter;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.net.SocketAddress;
+import javax.annotation.Nonnegative;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+
+public class TcpSocketAdapterImpl extends
AbstractClientSocketAdapter<Socket> implements TcpSocketAdapter {
+ public TcpSocketAdapterImpl(Socket socket) {
+ super(socket);
+ }
+
+ @Override
+ public void write(byte[] data) throws IOException {
+ this.write(data, this.socket.getRemoteSocketAddress());
+ }
+
+ @Override
+ public void write(byte[] data, SocketAddress remoteAddr) throws
IOException {
+ OutputStream outStream = this.socket.getOutputStream();
+
+ IOUtils.write(data, outStream);
+
+ outStream.flush();
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read() throws IOException {
+ return this.read(this.getProtocol().getDataSizeMax());
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read(@Nonnegative int bufferLen)
throws IOException {
+ return this.read(new byte[bufferLen]);
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read(byte[] buffer) throws
IOException {
+ int dataLen = IOUtils.read(this.socket.getInputStream(), buffer);
+
+ return new MutablePair<>(this.socket.getRemoteSocketAddress(),
ArrayUtils.subarray(buffer, 0, dataLen));
+ }
+
+ @Override
+ public void bind(SocketAddress bindSocketAddr) throws IOException {
+ this.socket.bind(bindSocketAddr);
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.socket.close();
+ }
+
+ @Override
+ public boolean isBound() {
+ return this.socket.isBound();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return this.socket.isClosed();
+ }
+
+ @Override
+ public InetProtocol getProtocol() {
+ return InetProtocol.TCP;
+ }
+
+ @Override
+ public int getReceiveBufferSize() throws IOException {
+ return this.socket.getReceiveBufferSize();
+ }
+
+ @Override
+ public void setReceiveBufferSize(@Nonnegative int receiveBufferSize)
throws IOException {
+ this.socket.setReceiveBufferSize(receiveBufferSize);
+ }
+
+ @Override
+ public boolean getReuseAddress() throws IOException {
+ return this.socket.getReuseAddress();
+ }
+
+ @Override
+ public void setReuseAddress(boolean reuseAddr) throws IOException {
+ this.socket.setReuseAddress(reuseAddr);
+ }
+
+ @Override
+ public int getSendBufferSize() throws IOException {
+ return this.socket.getSendBufferSize();
+ }
+
+ @Override
+ public void setSendBufferSize(@Nonnegative int sendBufferSize) throws
IOException {
+ this.socket.setSendBufferSize(sendBufferSize);
+ }
+
+ @Override
+ public Socket getSocket() {
+ return this.socket;
+ }
+
+ @Override
+ public int getTimeout() throws IOException {
+ return this.socket.getSoTimeout();
+ }
+
+ @Override
+ public void setTimeout(@Nonnegative int timeout) throws IOException {
+ this.socket.setSoTimeout(timeout);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/UdpSocketAdapterImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,112 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.UdpSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.utils.ToolSocketUtils;
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.SocketAddress;
+import javax.annotation.Nonnegative;
+import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+
+public class UdpSocketAdapterImpl extends
AbstractClientSocketAdapter<DatagramSocket> implements UdpSocketAdapter {
+ public UdpSocketAdapterImpl(DatagramSocket socket) {
+ super(socket);
+ }
+
+ @Override
+ public void write(byte[] data, SocketAddress remoteAddr) throws
IOException {
+ this.socket.send(ToolSocketUtils.createPacket(data, remoteAddr));
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read() throws IOException {
+ return this.read(this.getProtocol().getDataSizeMax());
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read(@Nonnegative int bufferLen)
throws IOException {
+ return this.read(new byte[bufferLen]);
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read(byte[] buffer) throws
IOException {
+ DatagramPacket packet = ToolSocketUtils.createPacket(buffer);
+
+ this.socket.receive(packet);
+
+ return new MutablePair<>(packet.getSocketAddress(),
packet.getData());
+ }
+
+ @Override
+ public void bind(SocketAddress bindSocketAddr) throws IOException {
+ this.socket.bind(bindSocketAddr);
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.socket.close();
+ }
+
+ @Override
+ public boolean isBound() {
+ return this.socket.isBound();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return this.socket.isClosed();
+ }
+
+ @Override
+ public InetProtocol getProtocol() {
+ return InetProtocol.UDP;
+ }
+
+ @Override
+ public int getReceiveBufferSize() throws IOException {
+ return this.socket.getReceiveBufferSize();
+ }
+
+ @Override
+ public void setReceiveBufferSize(@Nonnegative int receiveBufferSize)
throws IOException {
+ this.socket.setReceiveBufferSize(receiveBufferSize);
+ }
+
+ @Override
+ public boolean getReuseAddress() throws IOException {
+ return this.socket.getReuseAddress();
+ }
+
+ @Override
+ public void setReuseAddress(boolean reuseAddr) throws IOException {
+ this.socket.setReuseAddress(reuseAddr);
+ }
+
+ @Override
+ public int getSendBufferSize() throws IOException {
+ return this.socket.getSendBufferSize();
+ }
+
+ @Override
+ public void setSendBufferSize(@Nonnegative int sendBufferSize) throws
IOException {
+ this.socket.setSendBufferSize(sendBufferSize);
+ }
+
+ @Override
+ public DatagramSocket getSocket() {
+ return this.socket;
+ }
+
+ @Override
+ public int getTimeout() throws IOException {
+ return this.socket.getSoTimeout();
+ }
+
+ @Override
+ public void setTimeout(@Nonnegative int timeout) throws IOException {
+ this.socket.setSoTimeout(timeout);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/utils/ToolSocketUtils.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,19 @@
+package gov.hhs.onc.dcdt.net.sockets.utils;
+
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.SocketAddress;
+
+public abstract class ToolSocketUtils {
+ public static DatagramPacket createPacket(int bufferLen) throws
IOException {
+ return new DatagramPacket(new byte[bufferLen], bufferLen);
+ }
+
+ public static DatagramPacket createPacket(byte[] buffer) throws
IOException {
+ return new DatagramPacket(buffer, buffer.length);
+ }
+
+ public static DatagramPacket createPacket(byte[] buffer, SocketAddress
socketAddr) throws IOException {
+ return new DatagramPacket(buffer, buffer.length, socketAddr);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequest.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,13 @@
+package gov.hhs.onc.dcdt.service.dns.server;
+
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import javax.annotation.Nonnegative;
+
+public interface DnsServerRequest extends SocketRequest {
+ public boolean hasQuerySize();
+
+ public int getQuerySize();
+
+ @Nonnegative
+ public int setQuerySize(@Nonnegative int querySize);
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequestProcessingException.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,49 @@
+package gov.hhs.onc.dcdt.service.dns.server;
+
+import gov.hhs.onc.dcdt.service.dns.DnsServiceException;
+import javax.annotation.Nullable;
+import org.xbill.DNS.Message;
+
+public class DnsServerRequestProcessingException extends
DnsServiceException {
+ private final static long serialVersionUID = 0L;
+
+ private Message reqMsg;
+ private Message respMsg;
+
+ public DnsServerRequestProcessingException(@Nullable Message reqMsg,
@Nullable Message respMsg) {
+ this(reqMsg, respMsg, null, null);
+ }
+
+ public DnsServerRequestProcessingException(@Nullable Message reqMsg,
@Nullable Message respMsg, String msg) {
+ this(reqMsg, respMsg, msg, null);
+ }
+
+ public DnsServerRequestProcessingException(@Nullable Message reqMsg,
@Nullable Message respMsg, Throwable cause) {
+ this(reqMsg, respMsg, null, cause);
+ }
+
+ public DnsServerRequestProcessingException(@Nullable Message reqMsg,
@Nullable Message respMsg, String msg, Throwable cause) {
+ super(msg, cause);
+
+ this.reqMsg = reqMsg;
+ this.respMsg = respMsg;
+ }
+
+ public boolean hasRequestMessage() {
+ return this.reqMsg != null;
+ }
+
+ @Nullable
+ public Message getRequestMessage() {
+ return this.reqMsg;
+ }
+
+ public boolean hasResponseMessage() {
+ return this.respMsg != null;
+ }
+
+ @Nullable
+ public Message getResponseMessage() {
+ return this.respMsg;
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequestProcessor.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,6 @@
+package gov.hhs.onc.dcdt.service.dns.server;
+
+import gov.hhs.onc.dcdt.net.sockets.SocketRequestProcessor;
+
+public interface DnsServerRequestProcessor extends
SocketRequestProcessor<DnsServerRequest> {
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerTcpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,6 @@
+package gov.hhs.onc.dcdt.service.dns.server;
+
+import gov.hhs.onc.dcdt.net.sockets.TcpSocketListener;
+
+public interface DnsServerTcpSocketListener extends
TcpSocketListener<DnsServerRequest, DnsServerRequestProcessor> {
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerUdpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,6 @@
+package gov.hhs.onc.dcdt.service.dns.server;
+
+import gov.hhs.onc.dcdt.net.sockets.UdpSocketListener;
+
+public interface DnsServerUdpSocketListener extends
UdpSocketListener<DnsServerRequest, DnsServerRequestProcessor> {
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerRequestImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,38 @@
+package gov.hhs.onc.dcdt.service.dns.server.impl;
+
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.impl.AbstractSocketRequest;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequest;
+import gov.hhs.onc.dcdt.utils.ToolNumberUtils;
+import javax.annotation.Nonnegative;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+import org.xbill.DNS.Message;
+
+@Component("dnsServerReq")
+@Lazy
+@Scope("prototype")
+public class DnsServerRequestImpl extends AbstractSocketRequest implements
DnsServerRequest {
+ private int querySize = -1;
+
+ public DnsServerRequestImpl(InetProtocol protocol) {
+ super(protocol, Message.MAXLENGTH, Message.MAXLENGTH);
+ }
+
+ @Override
+ public boolean hasQuerySize() {
+ return ToolNumberUtils.isPositive(this.querySize);
+ }
+
+ @Override
+ public int getQuerySize() {
+ return this.querySize;
+ }
+
+ @Nonnegative
+ @Override
+ public int setQuerySize(@Nonnegative int querySize) {
+ return (this.querySize = querySize);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerRequestProcessorImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,100 @@
+package gov.hhs.onc.dcdt.service.dns.server.impl;
+
+import gov.hhs.onc.dcdt.config.instance.InstanceDnsConfig;
+import gov.hhs.onc.dcdt.dns.DnsException;
+import gov.hhs.onc.dcdt.dns.DnsMessageRcode;
+import gov.hhs.onc.dcdt.dns.DnsRecordType;
+import gov.hhs.onc.dcdt.dns.utils.ToolDnsMessageUtils;
+import gov.hhs.onc.dcdt.dns.utils.ToolDnsRecordUtils;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.impl.AbstractSocketRequestProcessor;
+import gov.hhs.onc.dcdt.service.dns.config.DnsServerConfig;
+import
gov.hhs.onc.dcdt.service.dns.server.DnsServerRequestProcessingException;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequest;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequestProcessor;
+import gov.hhs.onc.dcdt.utils.ToolClassUtils;
+import org.apache.commons.lang3.ArrayUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+import org.xbill.DNS.Message;
+import org.xbill.DNS.Name;
+import org.xbill.DNS.Record;
+
+@Component("dnsServerReqProcImpl")
+@Lazy
+@Scope("prototype")
+public class DnsServerRequestProcessorImpl extends
AbstractSocketRequestProcessor<DnsServerRequest> implements
DnsServerRequestProcessor {
+ private final static Logger LOGGER =
LoggerFactory.getLogger(DnsServerRequestProcessorImpl.class);
+
+ private DnsServerConfig serverConfig;
+
+ public DnsServerRequestProcessorImpl(DnsServerConfig serverConfig,
DnsServerRequest req) {
+ super(req);
+
+ this.serverConfig = serverConfig;
+ }
+
+ @Override
+ protected byte[] processError(byte[] reqData, Exception exception) {
+ return
ToolDnsMessageUtils.createErrorResponse(((DnsServerRequestProcessingException)
exception).getRequestMessage(), DnsMessageRcode.SERVFAIL).toWire();
+ }
+
+ @Override
+ protected byte[] processRequestInternal(byte[] reqData) throws
Exception {
+ InetProtocol protocol = this.req.getProtocol();
+ Message reqMsg = null, respMsg = null;
+ byte[] respData = null;
+
+ try {
+ respData = ToolDnsMessageUtils.toWire(protocol, (respMsg =
this.resolveQuery((reqMsg = ToolDnsMessageUtils.fromWire(protocol,
reqData)))));
+
+ LOGGER.trace(String.format("Resolved (class=%s) DNS server
query (protocol=%s, reqDataSize=%d, reqMsg={%s}): respDataSize=%d,
respMsg={%s}",
+ ToolClassUtils.getName(this), protocol.name(),
reqData.length, reqMsg, ArrayUtils.getLength(respData), respMsg));
+
+ return respData;
+ } catch (Exception e) {
+ throw new DnsServerRequestProcessingException(reqMsg, respMsg,
String.format(
+ "Unable to resolve (class=%s) DNS server query
(protocol=%s, reqDataSize=%d, reqMsg={%s}): respDataSize=%d, respMsg={%s}",
+ ToolClassUtils.getName(this), protocol.name(),
reqData.length, reqMsg, ArrayUtils.getLength(respData), respMsg), e);
+ }
+ }
+
+ private Message resolveQuery(Message reqMsg) throws DnsException {
+ Message respMsg = ToolDnsMessageUtils.createResponse(reqMsg);
+ Record questionRecord = reqMsg.getQuestion();
+
+ if (questionRecord == null) {
+ ToolDnsMessageUtils.setRcode(respMsg, DnsMessageRcode.FORMERR);
+
+ return respMsg;
+ }
+
+ DnsRecordType questionRecordType =
ToolDnsRecordUtils.findByType(questionRecord.getType());
+ Name questionName;
+
+ if (questionRecordType == null) {
+ ToolDnsMessageUtils.setRcode(respMsg, DnsMessageRcode.NXRRSET);
+
+ return respMsg;
+ } else if (!(questionName = questionRecord.getName()).isAbsolute()
|| questionName.isWild()) {
+ ToolDnsMessageUtils.setRcode(respMsg, DnsMessageRcode.REFUSED);
+
+ return respMsg;
+ }
+
+ InstanceDnsConfig authoritativeDnsConfig =
this.serverConfig.findAuthoritativeDnsConfig(questionRecord);
+
+ if (authoritativeDnsConfig != null) {
+ ToolDnsMessageUtils.setAnswers(respMsg,
authoritativeDnsConfig.findAnswers(questionRecord));
+ // noinspection ConstantConditions
+ ToolDnsMessageUtils.setAuthorities(respMsg, true,
authoritativeDnsConfig.getSoaRecordConfig().toRecord());
+ } else {
+ ToolDnsMessageUtils.setRcode(respMsg, DnsMessageRcode.REFUSED);
+ }
+
+ return respMsg;
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerTcpSocketListenerImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,68 @@
+package gov.hhs.onc.dcdt.service.dns.server.impl;
+
+import gov.hhs.onc.dcdt.beans.Phase;
+import gov.hhs.onc.dcdt.context.AutoStartup;
+import gov.hhs.onc.dcdt.dns.utils.ToolDnsMessageUtils;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.TcpServerSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.TcpSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.impl.AbstractTcpSocketListener;
+import gov.hhs.onc.dcdt.service.dns.config.DnsServerConfig;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequest;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequestProcessor;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerTcpSocketListener;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import javax.annotation.Resource;
+import org.apache.commons.lang3.ArrayUtils;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.context.annotation.Scope;
+import org.springframework.core.task.AsyncListenableTaskExecutor;
+import org.springframework.stereotype.Component;
+
+@AutoStartup(false)
+@Component("dnsServerSocketListenerTcp")
+@Lazy
+@Phase(Phase.PHASE_PRECEDENCE_HIGHEST)
+@Scope("prototype")
+public class DnsServerTcpSocketListenerImpl extends
AbstractTcpSocketListener<DnsServerRequest, DnsServerRequestProcessor>
implements
+ DnsServerTcpSocketListener {
+ private DnsServerConfig serverConfig;
+
+ public DnsServerTcpSocketListenerImpl(DnsServerConfig serverConfig) {
+ super(DnsServerRequest.class, DnsServerRequestProcessor.class,
serverConfig.toSocketAddress());
+
+ this.serverConfig = serverConfig;
+ }
+
+ @Override
+ protected DnsServerRequestProcessor createRequestProcessor(Object ...
reqProcArgs) {
+ return super.createRequestProcessor(ArrayUtils.add(reqProcArgs, 0,
this.serverConfig));
+ }
+
+ @Override
+ protected TcpSocketAdapter readRequest(TcpServerSocketAdapter
listenSocketAdapter, DnsServerRequest req) throws IOException {
+ TcpSocketAdapter reqSocketAdapter =
super.readRequest(listenSocketAdapter, req);
+ ByteBuffer reqBuffer = req.getRequestBuffer();
+ int reqQuerySize =
req.setQuerySize(ToolDnsMessageUtils.parseQuerySizeData(reqBuffer.get(0),
reqBuffer.get(1)));
+
+ reqBuffer.limit(reqBuffer.limit() + reqQuerySize);
+ reqBuffer.put(reqSocketAdapter.read(reqQuerySize).getRight());
+
+ return reqSocketAdapter;
+ }
+
+ @Override
+ protected DnsServerRequest createRequest(InetProtocol protocol) {
+ DnsServerRequest req = super.createRequest(protocol);
+
req.getRequestBuffer().limit(ToolDnsMessageUtils.DATA_SIZE_DNS_MSG_QUERY_SIZE_PREFIX);
+
+ return req;
+ }
+
+ @Override
+ @Resource(name = "taskExecServiceDnsServerReqQuery")
+ protected void setTaskExecutor(AsyncListenableTaskExecutor taskExec) {
+ super.setTaskExecutor(taskExec);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerUdpSocketListenerImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,42 @@
+package gov.hhs.onc.dcdt.service.dns.server.impl;
+
+import gov.hhs.onc.dcdt.beans.Phase;
+import gov.hhs.onc.dcdt.context.AutoStartup;
+import gov.hhs.onc.dcdt.net.sockets.impl.AbstractUdpSocketListener;
+import gov.hhs.onc.dcdt.service.dns.config.DnsServerConfig;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequest;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequestProcessor;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerUdpSocketListener;
+import javax.annotation.Resource;
+import org.apache.commons.lang3.ArrayUtils;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.context.annotation.Scope;
+import org.springframework.core.task.AsyncListenableTaskExecutor;
+import org.springframework.stereotype.Component;
+
+@AutoStartup(false)
+@Component("dnsServerSocketListenerUdp")
+@Lazy
+@Phase(Phase.PHASE_PRECEDENCE_HIGHEST)
+@Scope("prototype")
+public class DnsServerUdpSocketListenerImpl extends
AbstractUdpSocketListener<DnsServerRequest, DnsServerRequestProcessor>
implements
+ DnsServerUdpSocketListener {
+ private DnsServerConfig serverConfig;
+
+ public DnsServerUdpSocketListenerImpl(DnsServerConfig serverConfig) {
+ super(DnsServerRequest.class, DnsServerRequestProcessor.class,
serverConfig.toSocketAddress());
+
+ this.serverConfig = serverConfig;
+ }
+
+ @Override
+ protected DnsServerRequestProcessor createRequestProcessor(Object ...
reqProcArgs) {
+ return super.createRequestProcessor(ArrayUtils.add(reqProcArgs, 0,
this.serverConfig));
+ }
+
+ @Override
+ @Resource(name = "taskExecServiceDnsServerReqQuery")
+ protected void setTaskExecutor(AsyncListenableTaskExecutor taskExec) {
+ super.setTaskExecutor(taskExec);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/test/resources/META-INF/service/dns/service-dns-test.properties
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,4 @@
+#====================================================================================================
+# DNS SERVICE
+#====================================================================================================
+dcdt.test.func.service.dns.lookup.concurrent.threads=50
=======================================
--- /dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/ToolSocketOption.java Sat
Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,23 +0,0 @@
-package gov.hhs.onc.dcdt.net;
-
-import java.net.SocketOption;
-
-public class ToolSocketOption<T> implements SocketOption<T> {
- private String name;
- private Class<T> type;
-
- public ToolSocketOption(String name, Class<T> type) {
- this.name = name;
- this.type = type;
- }
-
- @Override
- public String name() {
- return this.name;
- }
-
- @Override
- public Class<T> type() {
- return this.type;
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListener.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,19 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import gov.hhs.onc.dcdt.beans.ToolBean;
-import java.io.IOException;
-import java.nio.channels.ByteChannel;
-import java.nio.channels.NetworkChannel;
-import java.nio.channels.SelectableChannel;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
-import javax.annotation.Nullable;
-import org.springframework.context.ApplicationContextAware;
-
-public interface ChannelListener<T extends SelectableChannel &
NetworkChannel, U extends SelectableChannel & ByteChannel & NetworkChannel,
V extends SelectableChannel & ByteChannel & NetworkChannel, W extends
SelectionAttachment, X extends ChannelListenerDataProcessor, Y extends
ChannelListenerDataProcessorCallback<X, W>>
- extends ApplicationContextAware, ToolBean {
- @Nullable
- public SelectionOperationType processSelection(SelectionKey selKey)
throws IOException;
-
- public void register(Selector selector) throws IOException;
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerDataProcessor.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,7 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import gov.hhs.onc.dcdt.beans.ToolBean;
-import java.util.concurrent.Callable;
-
-public interface ChannelListenerDataProcessor extends Callable<byte[]>,
ToolBean {
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerDataProcessorCallback.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,7 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import org.springframework.util.concurrent.ListenableFutureCallback;
-
-public interface ChannelListenerDataProcessorCallback<T extends
ChannelListenerDataProcessor, U extends SelectionAttachment> extends
- ListenableFutureCallback<byte[]> {
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerSelector.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,13 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import gov.hhs.onc.dcdt.beans.ToolLifecycleBean;
-import java.util.List;
-import javax.annotation.Nullable;
-
-public interface ChannelListenerSelector extends ToolLifecycleBean {
- public List<ChannelListener<?, ?, ?, ?, ?, ?>> getChannelListeners();
-
- public void setChannelListeners(@Nullable
ChannelListener<?, ?, ?, ?, ?, ?> ... channelListeners);
-
- public void setChannelListeners(@Nullable
Iterable<ChannelListener<?, ?, ?, ?, ?, ?>> channelListeners);
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelSocketOptions.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,8 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import gov.hhs.onc.dcdt.net.ToolSocketOption;
-import java.net.SocketOption;
-
-public interface ChannelSocketOptions {
- public final static SocketOption<Boolean> CHANNEL_BLOCKING = new
ToolSocketOption<>("CHANNEL_BLOCKING", Boolean.class);
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/SelectionAttachment.java
Tue Feb 25 21:16:43 2014 UTC
+++ /dev/null
@@ -1,26 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import java.net.SocketAddress;
-import java.nio.ByteBuffer;
-import javax.annotation.Nullable;
-
-public interface SelectionAttachment {
- public InetProtocol getProtocol();
-
- public void setProtocol(InetProtocol protocol);
-
- public ByteBuffer getRequestBuffer();
-
- public void setRequestBuffer(ByteBuffer reqBuffer);
-
- public ByteBuffer getResponseBuffer();
-
- public void setResponseBuffer(ByteBuffer respBuffer);
-
- @Nullable
- public SocketAddress getSocketAddress();
-
- @Nullable
- public SocketAddress setSocketAddress(@Nullable SocketAddress
socketAddr);
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/SelectionOperationType.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,25 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import java.nio.channels.SelectionKey;
-
-public enum SelectionOperationType {
- READ(SelectionKey.OP_READ), WRITE(SelectionKey.OP_WRITE),
ACCEPT(SelectionKey.OP_ACCEPT), CONNECT(SelectionKey.OP_CONNECT);
-
- private final int op;
-
- private SelectionOperationType(int op) {
- this.op = op;
- }
-
- public boolean isReady(SelectionKey selKey) {
- return this.isReady(selKey.readyOps());
- }
-
- public boolean isReady(int readyOps) {
- return (readyOps & this.op) != 0;
- }
-
- public int getOperation() {
- return this.op;
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListener.java
Tue Feb 25 21:16:43 2014 UTC
+++ /dev/null
@@ -1,208 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.beans.impl.AbstractToolBean;
-import gov.hhs.onc.dcdt.beans.utils.ToolBeanFactoryUtils;
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListener;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessor;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessorCallback;
-import gov.hhs.onc.dcdt.nio.channels.ChannelSocketOptions;
-import gov.hhs.onc.dcdt.nio.channels.SelectionAttachment;
-import gov.hhs.onc.dcdt.nio.channels.SelectionOperationType;
-import gov.hhs.onc.dcdt.nio.utils.ToolBufferUtils;
-import gov.hhs.onc.dcdt.nio.utils.ToolChannelUtils;
-import gov.hhs.onc.dcdt.utils.ToolArrayUtils;
-import gov.hhs.onc.dcdt.utils.ToolClassUtils;
-import gov.hhs.onc.dcdt.utils.ToolMapUtils;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.SocketOption;
-import java.net.StandardSocketOptions;
-import java.nio.ByteBuffer;
-import java.nio.channels.ByteChannel;
-import java.nio.channels.ClosedChannelException;
-import java.nio.channels.NetworkChannel;
-import java.nio.channels.SelectableChannel;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
-import java.util.List;
-import java.util.Map;
-import javax.annotation.Nullable;
-import org.apache.commons.collections4.map.LinkedMap;
-import org.apache.commons.lang3.tuple.MutablePair;
-import org.apache.commons.lang3.tuple.Pair;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.BeansException;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.support.AbstractApplicationContext;
-import org.springframework.core.task.AsyncListenableTaskExecutor;
-import org.springframework.util.concurrent.ListenableFutureCallback;
-import org.springframework.util.concurrent.ListenableFutureTask;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractChannelListener<T extends SelectableChannel
& NetworkChannel, U extends SelectableChannel & ByteChannel &
NetworkChannel, V extends SelectableChannel & ByteChannel & NetworkChannel,
W extends SelectionAttachment, X extends ChannelListenerDataProcessor, Y
extends ChannelListenerDataProcessorCallback<X, W>>
- extends AbstractToolBean implements ChannelListener<T, U, V, W, X, Y> {
- @SuppressWarnings({ "unchecked" })
- protected final static List<? extends Pair<? extends
SocketOption<?>, ?>> OPT_PAIRS_CHANNEL = ToolArrayUtils.asList(new
MutablePair<>(
- StandardSocketOptions.SO_REUSEADDR, true), new
MutablePair<>(ChannelSocketOptions.CHANNEL_BLOCKING, false));
-
- protected AbstractApplicationContext appContext;
- protected AsyncListenableTaskExecutor taskExec;
- protected Class<T> listenChannelClass;
- protected Class<U> readChannelClass;
- protected Class<V> writeChannelClass;
- protected Class<W> attachmentClass;
- protected Class<X> dataProcClass;
- protected Class<Y> dataProcCallbackClass;
- protected SelectionOperationType listenOpType;
- protected SelectionOperationType readOpType;
- protected SelectionOperationType writeOpType;
- protected InetProtocol protocol;
- protected InetSocketAddress bindSocketAddr;
-
- private final static Logger LOGGER =
LoggerFactory.getLogger(AbstractChannelListener.class);
-
- protected AbstractChannelListener(Class<T> listenChannelClass,
Class<U> readChannelClass, Class<V> writeChannelClass, Class<W>
attachmentClass,
- Class<X> dataProcClass, Class<Y> dataProcCallbackClass,
SelectionOperationType listenOpType, SelectionOperationType readOpType,
- SelectionOperationType writeOpType, InetProtocol protocol,
InetSocketAddress bindSocketAddr) {
- this.listenChannelClass = listenChannelClass;
- this.readChannelClass = readChannelClass;
- this.writeChannelClass = writeChannelClass;
- this.attachmentClass = attachmentClass;
- this.dataProcClass = dataProcClass;
- this.dataProcCallbackClass = dataProcCallbackClass;
- this.listenOpType = listenOpType;
- this.readOpType = readOpType;
- this.writeOpType = writeOpType;
- this.protocol = protocol;
- this.bindSocketAddr = bindSocketAddr;
- }
-
- @Nullable
- @Override
- public SelectionOperationType processSelection(SelectionKey selKey)
throws IOException {
- Pair<SelectionOperationType, ? extends Class<? extends
SelectableChannel>> selOp = this.getSelectedOperation(selKey);
-
- if (selOp != null) {
- SelectionOperationType selOpType = selOp.getLeft();
-
- W selAttachment =
this.attachmentClass.cast(selKey.attachment());
- boolean selProcessed;
-
- if (selOpType == this.readOpType) {
- selProcessed = this.processReadOperation(selKey,
this.readChannelClass.cast(selKey.channel()), selAttachment);
- } else if (selOpType == this.writeOpType) {
- selProcessed = this.processWriteOperation(selKey,
this.writeChannelClass.cast(selKey.channel()), selAttachment);
- } else {
- selProcessed = this.processListenOperation(selKey,
this.listenChannelClass.cast(selKey.channel()), selAttachment);
- }
-
- if (selProcessed) {
- LOGGER.trace(String.format("Channel listener (class=%s)
processed selection (opType=%s, channelClass=%s, attachmentClass=%s).",
- ToolClassUtils.getName(this), selOpType.name(),
ToolClassUtils.getName(selOp.getRight()),
ToolClassUtils.getName(selAttachment)));
- }
-
- return selOpType;
- } else {
- return null;
- }
- }
-
- @Override
- public void register(Selector selector) throws IOException {
- try {
- this.createListenChannel().register(selector,
this.listenOpType.getOperation(), this.createAttachment());
- } catch (ClosedChannelException ignored) {
- }
- }
-
- protected static boolean isOperationSelected(SelectionKey selKey,
SelectionOperationType selOpType, Class<? extends SelectableChannel>
selOpChannelClass) {
- return selOpType.isReady(selKey) &&
ToolClassUtils.isAssignable(selKey.channel().getClass(), selOpChannelClass);
- }
-
- protected boolean processWriteOperation(SelectionKey selKey, V
writeChannel, W selAttachment) throws IOException {
- selKey.attach(this.createAttachment());
- selKey.interestOps(this.readOpType.getOperation());
- selKey.selector().wakeup();
-
- return true;
- }
-
- protected boolean processReadOperation(SelectionKey selKey, U
readChannel, W selAttachment) throws IOException {
-
this.executeDataProcessorTask(ToolBufferUtils.flip(selAttachment.getRequestBuffer()),
this.createDataProcessorCallbacks(selKey, selAttachment));
-
- return true;
- }
-
- protected boolean processListenOperation(SelectionKey selKey, T
listenChannel, W selAttachment) throws IOException {
- selKey.selector().wakeup();
-
- return true;
- }
-
- protected void executeDataProcessorTask(ByteBuffer reqBuffer,
Iterable<? extends ListenableFutureCallback<byte[]>> dataProcCallbacks) {
- ListenableFutureTask<byte[]> dataProcFutureTask = new
ListenableFutureTask<>(this.createDataProcessor(this.protocol,
ToolBufferUtils.get(reqBuffer)));
-
- for (ListenableFutureCallback<byte[]> dataProcCallback :
dataProcCallbacks) {
- dataProcFutureTask.addCallback(dataProcCallback);
- }
-
- this.taskExec.execute(dataProcFutureTask);
- }
-
- @Nullable
- protected Pair<SelectionOperationType, ? extends Class<? extends
SelectableChannel>> getSelectedOperation(SelectionKey selKey) {
- SelectionOperationType selOpType;
- Class<? extends SelectableChannel> selChannelClass;
-
- return (isOperationSelected(selKey, (selOpType = this.readOpType),
(selChannelClass = this.readChannelClass))
- || isOperationSelected(selKey, (selOpType = this.writeOpType),
(selChannelClass = this.writeChannelClass)) || isOperationSelected(selKey,
- (selOpType = this.listenOpType), (selChannelClass =
this.listenChannelClass))) ? new MutablePair<>(selOpType,
selChannelClass) : null;
- }
-
- protected List<? extends Y> createDataProcessorCallbacks(SelectionKey
selKey, W selAttachment) {
- return ToolBeanFactoryUtils.createBeansOfType(this.appContext,
this.dataProcCallbackClass, this.readOpType, this.writeOpType,
this.protocol, selKey,
- selAttachment);
- }
-
- protected X createDataProcessor(Object ... beanCreationArgs) {
- return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.dataProcClass, beanCreationArgs);
- }
-
- protected W createAttachment() {
- return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.attachmentClass, this.protocol);
- }
-
- protected T createListenChannel() throws IOException {
- return
this.listenChannelClass.cast(this.initializeChannel(ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.listenChannelClass)).bind(
- this.bindSocketAddr));
- }
-
- protected <Z extends SelectableChannel & NetworkChannel> Z
initializeChannel(Z channel) throws IOException {
- Map<SocketOption<?>, Object> optMap = ToolMapUtils.putAll(new
LinkedMap<SocketOption<?>, Object>(), OPT_PAIRS_CHANNEL);
- Class<? extends SelectableChannel> channelClass =
channel.getClass();
- int bufferSize = this.protocol.getDataSizeMax();
-
- if (ToolClassUtils.isAssignable(channelClass,
this.readChannelClass)) {
- // noinspection ConstantConditions
- optMap.put(StandardSocketOptions.SO_RCVBUF, bufferSize);
- }
-
- if (ToolClassUtils.isAssignable(channelClass,
this.writeChannelClass)) {
- // noinspection ConstantConditions
- optMap.put(StandardSocketOptions.SO_SNDBUF, bufferSize);
- }
-
- return ToolChannelUtils.setOptions(channel, optMap);
- }
-
- @Override
- public void setApplicationContext(ApplicationContext appContext)
throws BeansException {
- this.appContext = (AbstractApplicationContext) appContext;
- }
-
- protected void setTaskExecutor(AsyncListenableTaskExecutor taskExec) {
- this.taskExec = taskExec;
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerDataProcessor.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,16 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.beans.impl.AbstractToolBean;
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessor;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractChannelListenerDataProcessor extends
AbstractToolBean implements ChannelListenerDataProcessor {
- protected InetProtocol protocol;
- protected byte[] reqData;
-
- protected AbstractChannelListenerDataProcessor(InetProtocol protocol,
byte[] reqData) {
- this.protocol = protocol;
- this.reqData = reqData;
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerDataProcessorCallback.java
Sat Feb 22 21:15:33 2014 UTC
+++ /dev/null
@@ -1,47 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessor;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessorCallback;
-import gov.hhs.onc.dcdt.nio.channels.SelectionAttachment;
-import gov.hhs.onc.dcdt.nio.channels.SelectionOperationType;
-import gov.hhs.onc.dcdt.nio.utils.ToolBufferUtils;
-import java.nio.channels.SelectionKey;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractChannelListenerDataProcessorCallback<T
extends ChannelListenerDataProcessor, U extends SelectionAttachment>
implements
- ChannelListenerDataProcessorCallback<T, U> {
- protected Class<T> dataProcClass;
- protected SelectionOperationType readOpType;
- protected SelectionOperationType writeOpType;
- protected InetProtocol protocol;
- protected SelectionKey selKey;
- protected U selAttachment;
-
- public AbstractChannelListenerDataProcessorCallback(Class<T>
dataProcClass, SelectionOperationType readOpType, SelectionOperationType
writeOpType,
- InetProtocol protocol, SelectionKey selKey, U selAttachment) {
- this.dataProcClass = dataProcClass;
- this.readOpType = readOpType;
- this.writeOpType = writeOpType;
- this.protocol = protocol;
- this.selKey = selKey;
- this.selAttachment = selAttachment;
- }
-
- @Override
- public void onSuccess(byte[] respData) {
- this.setResponseData(respData);
- }
-
- @Override
- public void onFailure(Throwable th) {
- }
-
- protected void setResponseData(byte[] respData) {
-
ToolBufferUtils.clear(this.selAttachment.getResponseBuffer()).put(respData);
-
- this.selKey.interestOps(this.writeOpType.getOperation());
-
- this.selKey.selector().wakeup();
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerSelector.java
Tue Feb 25 21:16:43 2014 UTC
+++ /dev/null
@@ -1,122 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.beans.impl.AbstractToolLifecycleBean;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListener;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerSelector;
-import gov.hhs.onc.dcdt.nio.channels.SelectionOperationType;
-import gov.hhs.onc.dcdt.utils.ToolArrayUtils;
-import gov.hhs.onc.dcdt.utils.ToolClassUtils;
-import gov.hhs.onc.dcdt.utils.ToolCollectionUtils;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.concurrent.Callable;
-import java.util.concurrent.FutureTask;
-import javax.annotation.Nullable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public abstract class AbstractChannelListenerSelector extends
AbstractToolLifecycleBean implements ChannelListenerSelector {
- protected class ChannelListenerSelectorDaemon implements
Callable<Void> {
- @Nullable
- @Override
- public Void call() throws Exception {
- Iterator<SelectionKey> selKeysIterator;
- SelectionKey selKey;
- SelectionOperationType selOpType;
-
- while (AbstractChannelListenerSelector.this.isRunning()) {
- if (AbstractChannelListenerSelector.this.selector.select()
== 0) {
- continue;
- }
-
- while ((selKeysIterator =
AbstractChannelListenerSelector.this.selector.selectedKeys().iterator()).hasNext()
- && ((selKey = selKeysIterator.next()) != null)) {
- selKeysIterator.remove();
-
- if (!selKey.isValid()) {
- continue;
- }
-
- selOpType = null;
-
- for (ChannelListener<?, ?, ?, ?, ?, ?>
channelListener : AbstractChannelListenerSelector.this.channelListeners) {
- if ((selOpType =
channelListener.processSelection(selKey)) != null) {
- break;
- }
- }
-
- if (selOpType == null) {
-
AbstractChannelListenerSelector.LOGGER.warn(String.format(
- "Channel listener selector (class=%s)
selection (readyOps=%d, channelClass=%s, attachmentClass=%s) not
processed.",
-
ToolClassUtils.getName(AbstractChannelListenerSelector.this),
selKey.readyOps(), ToolClassUtils.getName(selKey.channel()),
- ToolClassUtils.getName(selKey.attachment())));
-
- selKey.cancel();
- }
- }
- }
-
- return null;
- }
- }
-
- protected List<ChannelListener<?, ?, ?, ?, ?, ?>> channelListeners =
new ArrayList<>();
- protected Selector selector;
- protected FutureTask<Void> selectorDaemonTask;
-
- private final static Logger LOGGER =
LoggerFactory.getLogger(AbstractChannelListenerSelector.class);
-
- @Override
- public boolean isRunning() {
- return super.isRunning() && (this.selector != null) &&
this.selector.isOpen();
- }
-
- @Override
- protected void stopInternal() throws Exception {
- if ((this.selectorDaemonTask != null)
&& !this.selectorDaemonTask.isDone()) {
- this.selectorDaemonTask.cancel(true);
- }
-
- for (SelectionKey selKey : this.selector.keys()) {
- try {
- selKey.channel().close();
- } catch (Exception ignored) {
- }
- }
-
- try {
- this.selector.close();
- } catch (Exception ignored) {
- }
- }
-
- @Override
- protected void startInternal() throws Exception {
- this.selector = Selector.open();
-
- for (ChannelListener<?, ?, ?, ?, ?, ?> channelListener :
this.channelListeners) {
- channelListener.register(this.selector);
- }
-
- this.taskExec.execute((this.selectorDaemonTask = new
FutureTask<>(new ChannelListenerSelectorDaemon())));
- }
-
- @Override
- public List<ChannelListener<?, ?, ?, ?, ?, ?>> getChannelListeners() {
- return this.channelListeners;
- }
-
- @Override
- public void setChannelListeners(@Nullable
ChannelListener<?, ?, ?, ?, ?, ?> ... channelListeners) {
- this.setChannelListeners(ToolArrayUtils.asList(channelListeners));
- }
-
- @Override
- public void setChannelListeners(@Nullable
Iterable<ChannelListener<?, ?, ?, ?, ?, ?>> channelListeners) {
- this.channelListeners.clear();
- ToolCollectionUtils.addAll(this.channelListeners,
channelListeners);
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractSelectionAttachment.java
Tue Feb 25 21:16:43 2014 UTC
+++ /dev/null
@@ -1,64 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.SelectionAttachment;
-import java.net.SocketAddress;
-import java.nio.ByteBuffer;
-import javax.annotation.Nonnegative;
-import javax.annotation.Nullable;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractSelectionAttachment implements
SelectionAttachment {
- protected InetProtocol protocol;
- protected ByteBuffer reqBuffer;
- protected ByteBuffer respBuffer;
- protected SocketAddress socketAddr;
-
- protected AbstractSelectionAttachment(InetProtocol protocol,
@Nonnegative int reqBufferSize, @Nonnegative int respBufferSize) {
- this.protocol = protocol;
- this.reqBuffer = ByteBuffer.allocate(reqBufferSize);
- this.respBuffer = ByteBuffer.allocate(respBufferSize);
- }
-
- @Override
- public InetProtocol getProtocol() {
- return this.protocol;
- }
-
- @Override
- public void setProtocol(InetProtocol protocol) {
- this.protocol = protocol;
- }
-
- @Override
- public ByteBuffer getRequestBuffer() {
- return this.reqBuffer;
- }
-
- @Override
- public void setRequestBuffer(ByteBuffer reqBuffer) {
- this.reqBuffer = reqBuffer;
- }
-
- @Override
- public ByteBuffer getResponseBuffer() {
- return this.respBuffer;
- }
-
- @Override
- public void setResponseBuffer(ByteBuffer respBuffer) {
- this.respBuffer = respBuffer;
- }
-
- @Nullable
- @Override
- public SocketAddress getSocketAddress() {
- return this.socketAddr;
- }
-
- @Nullable
- @Override
- public SocketAddress setSocketAddress(@Nullable SocketAddress
socketAddr) {
- return (this.socketAddr = socketAddr);
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractTcpChannelListener.java
Mon May 5 03:47:29 2014 UTC
+++ /dev/null
@@ -1,76 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessor;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessorCallback;
-import gov.hhs.onc.dcdt.nio.channels.SelectionAttachment;
-import gov.hhs.onc.dcdt.nio.channels.SelectionOperationType;
-import gov.hhs.onc.dcdt.nio.utils.ToolBufferUtils;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.ServerSocketChannel;
-import java.nio.channels.SocketChannel;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractTcpChannelListener<T extends
SelectionAttachment, U extends ChannelListenerDataProcessor, V extends
ChannelListenerDataProcessorCallback<U, T>>
- extends AbstractChannelListener<ServerSocketChannel, SocketChannel,
SocketChannel, T, U, V> {
- protected AbstractTcpChannelListener(Class<T> attachmentClass,
Class<U> dataProcClass, Class<V> dataProcCallbackClass, InetSocketAddress
bindSocketAddr) {
- super(ServerSocketChannel.class, SocketChannel.class,
SocketChannel.class, attachmentClass, dataProcClass, dataProcCallbackClass,
- SelectionOperationType.ACCEPT, SelectionOperationType.READ,
SelectionOperationType.WRITE, InetProtocol.TCP, bindSocketAddr);
- }
-
- @Override
- protected boolean processWriteOperation(SelectionKey selKey,
SocketChannel writeChannel, T selAttachment) throws IOException {
- ByteBuffer respBuffer = selAttachment.getResponseBuffer();
-
- if (respBuffer.position() > 0) {
- writeChannel.write(ToolBufferUtils.flip(respBuffer));
-
- if (respBuffer.hasRemaining()) {
- respBuffer.compact();
-
- selKey.selector().wakeup();
-
- return false;
- }
- }
-
- return super.processWriteOperation(selKey, writeChannel,
selAttachment);
- }
-
- @Override
- protected boolean processReadOperation(SelectionKey selKey,
SocketChannel readChannel, T selAttachment) throws IOException {
- ByteBuffer reqBuffer = selAttachment.getRequestBuffer();
-
- if (readChannel.read(reqBuffer) == -1) {
- readChannel.close();
- selKey.cancel();
- } else if (!reqBuffer.hasRemaining()) {
- if (reqBuffer.position() > 0) {
- return super.processReadOperation(selKey, readChannel,
selAttachment);
- } else {
- selKey.cancel();
- }
- }
-
- selKey.selector().wakeup();
-
- return false;
- }
-
- @Override
- @SuppressWarnings({ "unchecked" })
- protected boolean processListenOperation(SelectionKey selKey,
ServerSocketChannel listenChannel, T selAttachment) throws IOException {
- SocketChannel readChannel = listenChannel.accept();
-
- if (readChannel != null) {
-
this.initializeChannel(readChannel).register(selKey.selector(),
this.readOpType.getOperation(), this.createAttachment());
-
- return super.processListenOperation(selKey, listenChannel,
selAttachment);
- } else {
- return false;
- }
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractUdpChannelListener.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,34 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessor;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessorCallback;
-import gov.hhs.onc.dcdt.nio.channels.SelectionAttachment;
-import gov.hhs.onc.dcdt.nio.channels.SelectionOperationType;
-import gov.hhs.onc.dcdt.nio.utils.ToolBufferUtils;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.nio.channels.DatagramChannel;
-import java.nio.channels.SelectionKey;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractUdpChannelListener<T extends
SelectionAttachment, U extends ChannelListenerDataProcessor, V extends
ChannelListenerDataProcessorCallback<U, T>>
- extends AbstractChannelListener<DatagramChannel, DatagramChannel,
DatagramChannel, T, U, V> {
- protected AbstractUdpChannelListener(Class<T> attachmentClass,
Class<U> dataProcClass, Class<V> dataProcCallbackClass, InetSocketAddress
bindSocketAddr) {
- super(DatagramChannel.class, DatagramChannel.class,
DatagramChannel.class, attachmentClass, dataProcClass,
dataProcCallbackClass,
- SelectionOperationType.READ, SelectionOperationType.READ,
SelectionOperationType.WRITE, InetProtocol.UDP, bindSocketAddr);
- }
-
- @Override
- protected boolean processWriteOperation(SelectionKey selKey,
DatagramChannel writeChannel, T selAttachment) throws IOException {
-
writeChannel.send(ToolBufferUtils.flip(selAttachment.getResponseBuffer()),
selAttachment.getSocketAddress());
-
- return super.processWriteOperation(selKey, writeChannel,
selAttachment);
- }
-
- @Override
- protected boolean processReadOperation(SelectionKey selKey,
DatagramChannel readChannel, T selAttachment) throws IOException {
- return
(selAttachment.setSocketAddress(readChannel.receive(selAttachment.getRequestBuffer())) !=
null)
- && (selAttachment.getRequestBuffer().position() > 0) &&
super.processReadOperation(selKey, readChannel, selAttachment);
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/utils/ToolChannelUtils.java
Sat Feb 22 21:15:33 2014 UTC
+++ /dev/null
@@ -1,35 +0,0 @@
-package gov.hhs.onc.dcdt.nio.utils;
-
-import gov.hhs.onc.dcdt.net.ToolSocketOption;
-import gov.hhs.onc.dcdt.nio.channels.ChannelSocketOptions;
-import gov.hhs.onc.dcdt.utils.ToolClassUtils;
-import gov.hhs.onc.dcdt.utils.ToolMapUtils;
-import java.io.IOException;
-import java.net.SocketOption;
-import java.nio.channels.NetworkChannel;
-import java.nio.channels.SelectableChannel;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Map.Entry;
-
-public abstract class ToolChannelUtils {
- @SuppressWarnings({ "unchecked" })
- public static <T extends SelectableChannel & NetworkChannel> T
setOption(T channel, Entry<? extends SocketOption<?>, ?> optEntry) throws
IOException {
- return setOptions(channel, ToolMapUtils.putAll(new
HashMap<SocketOption<?>, Object>(1), optEntry));
- }
-
- @SuppressWarnings({ "unchecked" })
- public static <T extends SelectableChannel & NetworkChannel> T
setOptions(T channel, Map<? extends SocketOption<?>, ?> optMap) throws
IOException {
- for (SocketOption<?> opt : optMap.keySet()) {
- if (ToolClassUtils.isAssignable(opt.getClass(),
ToolSocketOption.class)) {
- if (opt == ChannelSocketOptions.CHANNEL_BLOCKING) {
-
channel.configureBlocking(ChannelSocketOptions.CHANNEL_BLOCKING.type().cast(optMap.get(opt)));
- }
- } else {
- channel.setOption(((SocketOption<Object>) opt),
optMap.get(opt));
- }
- }
-
- return channel;
- }
-}
=======================================
--- /dcdt-core/src/main/resources/META-INF/spring/spring-core-nio.xml Sat
Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,30 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<beans:beans
- xmlns:aop="http://www.springframework.org/schema/aop"
- xmlns:beans="http://www.springframework.org/schema/beans"
- xmlns:c="http://www.springframework.org/schema/c"
- xmlns:context="http://www.springframework.org/schema/context"
- xmlns:dcdt="http://direct-test.com/spring/schema/spring-dcdt"
- xmlns:p="http://www.springframework.org/schema/p"
- xmlns:task="http://www.springframework.org/schema/task"
- xmlns:util="http://www.springframework.org/schema/util"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="
- http://direct-test.com/spring/schema/spring-dcdt
http://direct-test.com/spring/schema/spring-dcdt.xsd
- http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop.xsd
- http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
- http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
- http://www.springframework.org/schema/task
http://www.springframework.org/schema/task/spring-task.xsd
- http://www.springframework.org/schema/util
http://www.springframework.org/schema/util/spring-util.xsd">
-
-
<!--====================================================================================================
- = CHANNELS
-
=====================================================================================================-->
- <beans:bean id="channel" class="java.nio.channels.SelectableChannel"
abstract="true"/>
-
- <beans:bean id="channelUdp" class="java.nio.channels.DatagramChannel"
parent="channel" factory-method="open" lazy-init="true" scope="prototype"/>
-
- <beans:bean id="channelTcp" class="java.nio.channels.SocketChannel"
parent="channel" factory-method="open" lazy-init="true" scope="prototype"/>
-
- <beans:bean id="channelTcpListen"
class="java.nio.channels.ServerSocketChannel" parent="channel"
factory-method="open" lazy-init="true" scope="prototype"/>
-</beans:beans>
=======================================
***Additional files exist in this changeset.***
Branch: default
Author: Michal Kotelba <michal....@esacinc.com>
Date: Thu May 22 06:39:17 2014 UTC
Log: - Supports DCDT-226.
- Refactored DNS service + underlying (N)IO framework to utilize adapted
sockets instead of NIO channels.
- Added DNS service concurrent load functional testing.
http://code.google.com/p/direct-certificate-discovery-tool/source/detail?r=d0a260ccd4e4
Added:
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/ClientSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketAdapterOptions.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketRequest.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketRequestProcessor.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpServerSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpSocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/ToolSocketOption.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/UdpSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/UdpSocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractClientSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketAdapter.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketRequest.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketRequestProcessor.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractTcpSocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractUdpSocketListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/TcpServerSocketAdapterImpl.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/TcpSocketAdapterImpl.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/UdpSocketAdapterImpl.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/utils/ToolSocketUtils.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequest.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequestProcessingException.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequestProcessor.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerTcpSocketListener.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerUdpSocketListener.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerRequestImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerRequestProcessorImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerTcpSocketListenerImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerUdpSocketListenerImpl.java
/dcdt-service-dns/src/test/resources/META-INF/service/dns/service-dns-test.properties
Deleted:
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/ToolSocketOption.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerDataProcessor.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerDataProcessorCallback.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerSelector.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelSocketOptions.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/SelectionAttachment.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/SelectionOperationType.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerDataProcessor.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerDataProcessorCallback.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerSelector.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractSelectionAttachment.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractTcpChannelListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractUdpChannelListener.java
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/utils/ToolChannelUtils.java
/dcdt-core/src/main/resources/META-INF/spring/spring-core-nio.xml
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerChannelListenerSelector.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerQueryAttachment.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerQueryResolutionException.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerQueryResolver.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerQueryResolverCallback.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerChannelListenerSelectorImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerQueryAttachmentImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerQueryResolverCallbackImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerQueryResolverImpl.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerTcpChannelListener.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerUdpChannelListener.java
Modified:
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/ToolNetException.java
/dcdt-core/src/main/resources/META-INF/spring/spring-core-net.xml
/dcdt-core/src/test/java/gov/hhs/onc/dcdt/config/instance/InstanceConfigRegistryFunctionalTests.java
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerImpl.java
/dcdt-service-dns/src/main/resources/META-INF/spring/spring-service-dns.xml
/dcdt-service-dns/src/test/java/gov/hhs/onc/dcdt/service/dns/DnsServiceFunctionalTests.java
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/ClientSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,22 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.SocketAddress;
+import javax.annotation.Nonnegative;
+import org.apache.commons.lang3.tuple.Pair;
+
+public interface ClientSocketAdapter<T extends Closeable> extends
SocketAdapter<T> {
+ public void write(byte[] data, SocketAddress remoteAddr) throws
IOException;
+
+ public Pair<SocketAddress, byte[]> read() throws IOException;
+
+ public Pair<SocketAddress, byte[]> read(@Nonnegative int bufferLen)
throws IOException;
+
+ public Pair<SocketAddress, byte[]> read(byte[] buffer) throws
IOException;
+
+ @Nonnegative
+ public int getSendBufferSize() throws IOException;
+
+ public void setSendBufferSize(@Nonnegative int sendBufferSize) throws
IOException;
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,40 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import gov.hhs.onc.dcdt.beans.ToolBean;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.net.SocketOption;
+import java.util.Map;
+import javax.annotation.Nonnegative;
+
+public interface SocketAdapter<T extends Closeable> extends Closeable,
ToolBean {
+ public void bind(SocketAddress bindSocketAddr) throws IOException;
+
+ public void setOptions(Map<? extends SocketOption<?>, ?> optMap)
throws IOException;
+
+ public <U> void setOption(SocketOption<U> optKey, U optValue) throws
IOException;
+
+ public boolean isBound();
+
+ public boolean isClosed();
+
+ public InetProtocol getProtocol();
+
+ @Nonnegative
+ public int getReceiveBufferSize() throws IOException;
+
+ public void setReceiveBufferSize(@Nonnegative int receiveBufferSize)
throws IOException;
+
+ public boolean getReuseAddress() throws IOException;
+
+ public void setReuseAddress(boolean reuseAddr) throws IOException;
+
+ public T getSocket();
+
+ @Nonnegative
+ public int getTimeout() throws IOException;
+
+ public void setTimeout(@Nonnegative int timeout) throws IOException;
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketAdapterOptions.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,10 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.net.SocketOption;
+
+public final class SocketAdapterOptions {
+ public final static SocketOption<Integer> TIMEOUT = new
ToolSocketOption<>("SO_TIMEOUT", Integer.class);
+
+ private SocketAdapterOptions() {
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,9 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import gov.hhs.onc.dcdt.beans.ToolLifecycleBean;
+import java.io.Closeable;
+import org.springframework.context.ApplicationContextAware;
+
+public interface SocketListener<T extends Closeable, U extends
SocketAdapter<T>, V extends Closeable, W extends ClientSocketAdapter<V>, X
extends SocketRequest, Y extends SocketRequestProcessor<X>>
+ extends ApplicationContextAware, ToolLifecycleBean {
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketRequest.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,20 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import gov.hhs.onc.dcdt.beans.ToolBean;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import javax.annotation.Nullable;
+
+public interface SocketRequest extends ToolBean {
+ public InetProtocol getProtocol();
+
+ @Nullable
+ public SocketAddress getRemoteAddress();
+
+ public void setRemoteAddress(@Nullable SocketAddress remoteAddr);
+
+ public ByteBuffer getRequestBuffer();
+
+ public ByteBuffer getResponseBuffer();
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/SocketRequestProcessor.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,7 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import gov.hhs.onc.dcdt.beans.ToolBean;
+
+public interface SocketRequestProcessor<T extends SocketRequest> extends
ToolBean {
+ public byte[] processRequest();
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpServerSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,9 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+
+public interface TcpServerSocketAdapter extends
SocketAdapter<ServerSocket> {
+ public Socket accept() throws IOException;
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,8 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.io.IOException;
+import java.net.Socket;
+
+public interface TcpSocketAdapter extends ClientSocketAdapter<Socket> {
+ public void write(byte[] data) throws IOException;
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/TcpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,8 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+
+public interface TcpSocketListener<T extends SocketRequest, U extends
SocketRequestProcessor<T>> extends
+ SocketListener<ServerSocket, TcpServerSocketAdapter, Socket,
TcpSocketAdapter, T, U> {
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/ToolSocketOption.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,23 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.net.SocketOption;
+
+public class ToolSocketOption<T> implements SocketOption<T> {
+ private String name;
+ private Class<T> type;
+
+ public ToolSocketOption(String name, Class<T> type) {
+ this.name = name;
+ this.type = type;
+ }
+
+ @Override
+ public String name() {
+ return this.name;
+ }
+
+ @Override
+ public Class<T> type() {
+ return this.type;
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/UdpSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,6 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.net.DatagramSocket;
+
+public interface UdpSocketAdapter extends
ClientSocketAdapter<DatagramSocket> {
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/UdpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,7 @@
+package gov.hhs.onc.dcdt.net.sockets;
+
+import java.net.DatagramSocket;
+
+public interface UdpSocketListener<T extends SocketRequest, U extends
SocketRequestProcessor<T>> extends
+ SocketListener<DatagramSocket, UdpSocketAdapter, DatagramSocket,
UdpSocketAdapter, T, U> {
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractClientSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,22 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.sockets.ClientSocketAdapter;
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.SocketOption;
+import java.net.StandardSocketOptions;
+
+public abstract class AbstractClientSocketAdapter<T extends Closeable>
extends AbstractSocketAdapter<T> implements ClientSocketAdapter<T> {
+ protected AbstractClientSocketAdapter(T socket) {
+ super(socket);
+ }
+
+ @Override
+ protected <U> void setOptionInternal(SocketOption<U> optKey, U
optValue) throws IOException {
+ if (optKey == StandardSocketOptions.SO_SNDBUF) {
+ this.setSendBufferSize(((Integer) optValue));
+ } else {
+ super.setOptionInternal(optKey, optValue);
+ }
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketAdapter.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,54 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.beans.impl.AbstractToolBean;
+import gov.hhs.onc.dcdt.net.sockets.SocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.SocketAdapterOptions;
+import gov.hhs.onc.dcdt.net.ToolNetException;
+import gov.hhs.onc.dcdt.utils.ToolClassUtils;
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.SocketOption;
+import java.net.StandardSocketOptions;
+import java.util.Map;
+import java.util.Map.Entry;
+
+public abstract class AbstractSocketAdapter<T extends Closeable> extends
AbstractToolBean implements SocketAdapter<T> {
+ protected T socket;
+
+ protected AbstractSocketAdapter(T socket) {
+ this.socket = socket;
+ }
+
+ @Override
+ @SuppressWarnings({ "unchecked" })
+ public void setOptions(Map<? extends SocketOption<?>, ?> optMap)
throws IOException {
+ for (Entry<? extends SocketOption<?>, ?> optEntry :
optMap.entrySet()) {
+ this.setOption(((SocketOption<Object>) optEntry.getKey()),
optEntry.getValue());
+ }
+ }
+
+ @Override
+ public <U> void setOption(SocketOption<U> optKey, U optValue) throws
IOException {
+ try {
+ this.setOptionInternal(optKey, optValue);
+ } catch (ToolNetException e) {
+ throw e;
+ } catch (IOException e) {
+ throw new ToolNetException(String.format("Unable to set socket
(class=%s) adapter (class=%s) option (key=%s) value: %s",
+ ToolClassUtils.getName(this.socket),
ToolClassUtils.getName(this), optKey, optValue), e);
+ }
+ }
+
+ protected <U> void setOptionInternal(SocketOption<U> optKey, U
optValue) throws IOException {
+ if (optKey == StandardSocketOptions.SO_RCVBUF) {
+ this.setReceiveBufferSize(((Integer) optValue));
+ } else if (optKey == StandardSocketOptions.SO_REUSEADDR) {
+ this.setReuseAddress(((Boolean) optValue));
+ } else if (optKey == SocketAdapterOptions.TIMEOUT) {
+ this.setTimeout(((Integer) optValue));
+ } else {
+ throw new ToolNetException(String.format("Unable to set
unknown socket (class=%s) adapter (class=%s) option (key=%s) value: %s",
+ ToolClassUtils.getName(this.socket),
ToolClassUtils.getName(this), optKey, optValue));
+ }
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,181 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.beans.impl.AbstractToolLifecycleBean;
+import gov.hhs.onc.dcdt.beans.utils.ToolBeanFactoryUtils;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.ClientSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.SocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.SocketListener;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequestProcessor;
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.FutureTask;
+import javax.annotation.Nullable;
+import org.apache.mina.util.ConcurrentHashSet;
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.util.concurrent.ListenableFutureCallback;
+import org.springframework.util.concurrent.ListenableFutureTask;
+
+public abstract class AbstractSocketListener<T extends Closeable, U
extends SocketAdapter<T>, V extends Closeable, W extends
ClientSocketAdapter<V>, X extends SocketRequest, Y extends
SocketRequestProcessor<X>>
+ extends AbstractToolLifecycleBean implements SocketListener<T, U, V,
W, X, Y> {
+ protected class SocketListenDaemon implements Callable<Void> {
+ @Nullable
+ @Override
+ public Void call() throws Exception {
+ while ((AbstractSocketListener.this.listenSocketAdapter !=
null) && !AbstractSocketListener.this.listenSocketAdapter.isClosed()) {
+ X req =
AbstractSocketListener.this.createRequest(AbstractSocketListener.this.listenSocketAdapter.getProtocol());
+
+ final W reqSocketAdapter =
AbstractSocketListener.this.readRequest(AbstractSocketListener.this.listenSocketAdapter,
req);
+
AbstractSocketListener.this.reqSocketAdapters.add(reqSocketAdapter);
+
+ final ListenableFutureTask<Void> reqDaemonTask =
AbstractSocketListener.this.createRequestDaemonTask(reqSocketAdapter, req);
+
AbstractSocketListener.this.reqDaemonTasks.add(reqDaemonTask);
+
+ reqDaemonTask.addCallback(new
ListenableFutureCallback<Void>() {
+ @Override
+ public void onSuccess(@Nullable Void result) {
+ this.processResponse();
+ }
+
+ @Override
+ public void onFailure(Throwable th) {
+ this.processResponse();
+ }
+
+ private void processResponse() {
+
AbstractSocketListener.this.reqDaemonTasks.remove(reqDaemonTask);
+
AbstractSocketListener.this.reqSocketAdapters.remove(reqSocketAdapter);
+ }
+ });
+
+
AbstractSocketListener.this.taskExec.execute(reqDaemonTask);
+ }
+
+ return null;
+ }
+ }
+
+ protected class SocketRequestDaemon implements Callable<Void> {
+ protected W reqSocketAdapter;
+ protected X req;
+
+ protected SocketRequestDaemon(W reqSocketAdapter, X req) {
+ this.reqSocketAdapter = reqSocketAdapter;
+ this.req = req;
+ }
+
+ @Nullable
+ @Override
+ public Void call() throws Exception {
+
AbstractSocketListener.this.writeResponse(this.reqSocketAdapter,
AbstractSocketListener.this.createRequestProcessor(req).processRequest(),
+ req.getRemoteAddress());
+
+ return null;
+ }
+ }
+
+ protected Class<T> listenSocketClass;
+ protected Class<U> listenSocketAdapterClass;
+ protected Class<V> reqSocketClass;
+ protected Class<W> reqSocketAdapterClass;
+ protected Class<X> reqClass;
+ protected Class<Y> reqProcClass;
+ protected SocketAddress bindSocketAddr;
+ protected AbstractApplicationContext appContext;
+ protected U listenSocketAdapter;
+ protected Set<W> reqSocketAdapters = new ConcurrentHashSet<>();
+ protected ListenableFutureTask<Void> listenDaemonTask;
+ protected Set<ListenableFutureTask<Void>> reqDaemonTasks = new
ConcurrentHashSet<>();
+
+ protected AbstractSocketListener(Class<T> listenSocketClass, Class<U>
listenSocketAdapterClass, Class<V> reqSocketClass, Class<W>
reqSocketAdapterClass,
+ Class<X> reqClass, Class<Y> reqProcClass, SocketAddress
bindSocketAddr) {
+ this.listenSocketClass = listenSocketClass;
+ this.listenSocketAdapterClass = listenSocketAdapterClass;
+ this.reqSocketClass = reqSocketClass;
+ this.reqSocketAdapterClass = reqSocketAdapterClass;
+ this.reqClass = reqClass;
+ this.reqProcClass = reqProcClass;
+ this.bindSocketAddr = bindSocketAddr;
+ }
+
+ @Override
+ public boolean isRunning() {
+ return (super.isRunning() && (((this.listenSocketAdapter != null)
&& !listenSocketAdapter.isClosed()) || !this.reqSocketAdapters.isEmpty()
+ || ((this.listenDaemonTask != null)
&& !this.listenDaemonTask.isDone()) || !this.reqDaemonTasks.isEmpty()));
+ }
+
+ @Override
+ protected void stopInternal() throws Exception {
+ if ((this.listenDaemonTask != null)
&& !this.listenDaemonTask.isDone()) {
+ this.listenDaemonTask.cancel(true);
+ }
+
+ if ((this.listenSocketAdapter != null)
&& !this.listenSocketAdapter.isClosed()) {
+ this.listenSocketAdapter.close();
+ }
+
+ for (FutureTask<Void> reqDaemonTask : this.reqDaemonTasks) {
+ if (!reqDaemonTask.isDone()) {
+ reqDaemonTask.cancel(true);
+ }
+ }
+
+ for (W reqSocketAdapter : this.reqSocketAdapters) {
+ if (!reqSocketAdapter.isClosed()) {
+ reqSocketAdapter.close();
+ }
+ }
+ }
+
+ @Override
+ protected void startInternal() throws Exception {
+ (this.listenSocketAdapter =
this.createListenSocketAdapter(this.createListenSocket())).bind(this.bindSocketAddr);
+
+ this.taskExec.execute((this.listenDaemonTask =
this.createListenDaemonTask()));
+ }
+
+ protected void writeResponse(W reqSocketAdapter, byte[] respData,
SocketAddress remoteAddr) throws IOException {
+ reqSocketAdapter.write(respData, remoteAddr);
+ }
+
+ protected Y createRequestProcessor(Object ... reqProcArgs) {
+ return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.reqProcClass, reqProcArgs);
+ }
+
+ protected abstract W readRequest(U listenSocketAdapter, X req) throws
IOException;
+
+ protected X createRequest(InetProtocol protocol) {
+ return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.reqClass, protocol);
+ }
+
+ protected ListenableFutureTask<Void> createRequestDaemonTask(W
respSocketAdapter, X req) {
+ return new ListenableFutureTask<>(new
SocketRequestDaemon(respSocketAdapter, req));
+ }
+
+ protected ListenableFutureTask<Void> createListenDaemonTask() {
+ return new ListenableFutureTask<>(new SocketListenDaemon());
+ }
+
+ protected W createRequestSocketAdapter(V reqSocket) throws IOException
{
+ return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.reqSocketAdapterClass, reqSocket);
+ }
+
+ protected U createListenSocketAdapter(T listenSocket) throws
IOException {
+ return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.listenSocketAdapterClass, listenSocket);
+ }
+
+ protected T createListenSocket(Object ... listenSocketArgs) {
+ return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.listenSocketClass, listenSocketArgs);
+ }
+
+ @Override
+ public void setApplicationContext(ApplicationContext appContext)
throws BeansException {
+ this.appContext = ((AbstractApplicationContext) appContext);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketRequest.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,48 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.beans.impl.AbstractToolBean;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import javax.annotation.Nonnegative;
+import javax.annotation.Nullable;
+
+public abstract class AbstractSocketRequest extends AbstractToolBean
implements SocketRequest {
+ protected InetProtocol protocol;
+ protected SocketAddress remoteAddr;
+ protected ByteBuffer reqBuffer;
+ protected ByteBuffer respBuffer;
+
+ protected AbstractSocketRequest(InetProtocol protocol, @Nonnegative
int reqBufferSize, @Nonnegative int respBufferSize) {
+ this.protocol = protocol;
+ this.reqBuffer = ByteBuffer.allocate(reqBufferSize);
+ this.respBuffer = ByteBuffer.allocate(respBufferSize);
+ }
+
+ @Override
+ public InetProtocol getProtocol() {
+ return this.protocol;
+ }
+
+ @Nullable
+ @Override
+ public SocketAddress getRemoteAddress() {
+ return this.remoteAddr;
+ }
+
+ @Override
+ public void setRemoteAddress(@Nullable SocketAddress remoteAddr) {
+ this.remoteAddr = remoteAddr;
+ }
+
+ @Override
+ public ByteBuffer getRequestBuffer() {
+ return this.reqBuffer;
+ }
+
+ @Override
+ public ByteBuffer getResponseBuffer() {
+ return this.respBuffer;
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractSocketRequestProcessor.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,33 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.beans.impl.AbstractToolBean;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequestProcessor;
+import gov.hhs.onc.dcdt.nio.utils.ToolBufferUtils;
+import java.nio.ByteBuffer;
+
+public abstract class AbstractSocketRequestProcessor<T extends
SocketRequest> extends AbstractToolBean implements
SocketRequestProcessor<T> {
+ protected T req;
+
+ protected AbstractSocketRequestProcessor(T req) {
+ this.req = req;
+ }
+
+ @Override
+ public byte[] processRequest() {
+ byte[] reqData =
ToolBufferUtils.get(ToolBufferUtils.flip(this.req.getRequestBuffer()));
+ ByteBuffer respBuffer = this.req.getResponseBuffer();
+
+ try {
+ respBuffer.put(this.processRequestInternal(reqData));
+ } catch (Exception e) {
+ respBuffer.put(this.processError(reqData, e));
+ }
+
+ return ToolBufferUtils.get(ToolBufferUtils.flip(respBuffer));
+ }
+
+ protected abstract byte[] processError(byte[] reqData, Exception
exception);
+
+ protected abstract byte[] processRequestInternal(byte[] reqData)
throws Exception;
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractTcpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,33 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequestProcessor;
+import gov.hhs.onc.dcdt.net.sockets.TcpServerSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.TcpSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.TcpSocketListener;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import org.apache.commons.lang3.tuple.Pair;
+
+public class AbstractTcpSocketListener<T extends SocketRequest, U extends
SocketRequestProcessor<T>> extends
+ AbstractSocketListener<ServerSocket, TcpServerSocketAdapter, Socket,
TcpSocketAdapter, T, U> implements TcpSocketListener<T, U> {
+ protected AbstractTcpSocketListener(Class<T> reqClass, Class<U>
reqProcClass, SocketAddress bindSocketAddr) {
+ super(ServerSocket.class, TcpServerSocketAdapter.class,
Socket.class, TcpSocketAdapter.class, reqClass, reqProcClass,
bindSocketAddr);
+ }
+
+ @Override
+ protected TcpSocketAdapter readRequest(TcpServerSocketAdapter
listenSocketAdapter, T req) throws IOException {
+ TcpSocketAdapter reqSocketAdapter =
this.createRequestSocketAdapter(listenSocketAdapter.accept());
+ ByteBuffer reqBuffer = req.getRequestBuffer();
+ Pair<SocketAddress, byte[]> reqPair =
reqSocketAdapter.read(reqBuffer.remaining());
+
+ req.setRemoteAddress(reqPair.getLeft());
+
+ reqBuffer.put(reqPair.getRight());
+
+ return reqSocketAdapter;
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/AbstractUdpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,35 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import gov.hhs.onc.dcdt.net.sockets.SocketRequestProcessor;
+import gov.hhs.onc.dcdt.net.sockets.UdpSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.UdpSocketListener;
+import java.io.IOException;
+import java.net.DatagramSocket;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import org.apache.commons.lang3.tuple.Pair;
+
+public abstract class AbstractUdpSocketListener<T extends SocketRequest, U
extends SocketRequestProcessor<T>> extends
+ AbstractSocketListener<DatagramSocket, UdpSocketAdapter,
DatagramSocket, UdpSocketAdapter, T, U> implements UdpSocketListener<T, U> {
+ protected AbstractUdpSocketListener(Class<T> reqClass, Class<U>
reqProcClass, SocketAddress bindSocketAddr) {
+ super(DatagramSocket.class, UdpSocketAdapter.class,
DatagramSocket.class, UdpSocketAdapter.class, reqClass, reqProcClass,
bindSocketAddr);
+ }
+
+ @Override
+ protected UdpSocketAdapter readRequest(UdpSocketAdapter
listenSocketAdapter, T req) throws IOException {
+ ByteBuffer reqBuffer = req.getRequestBuffer();
+ Pair<SocketAddress, byte[]> reqPair =
listenSocketAdapter.read(reqBuffer.remaining());
+
+ req.setRemoteAddress(reqPair.getLeft());
+
+ reqBuffer.put(reqPair.getRight());
+
+ return listenSocketAdapter;
+ }
+
+ @Override
+ protected DatagramSocket createListenSocket(Object ...
listenSocketArgs) {
+ return super.createListenSocket(((SocketAddress) null));
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/TcpServerSocketAdapterImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,80 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.TcpServerSocketAdapter;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketAddress;
+import javax.annotation.Nonnegative;
+
+public class TcpServerSocketAdapterImpl extends
AbstractSocketAdapter<ServerSocket> implements TcpServerSocketAdapter {
+ public TcpServerSocketAdapterImpl(ServerSocket socket) {
+ super(socket);
+ }
+
+ @Override
+ public Socket accept() throws IOException {
+ return this.socket.accept();
+ }
+
+ @Override
+ public void bind(SocketAddress bindSocketAddr) throws IOException {
+ this.socket.bind(bindSocketAddr);
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.socket.close();
+ }
+
+ @Override
+ public boolean isBound() {
+ return this.socket.isBound();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return this.socket.isClosed();
+ }
+
+ @Override
+ public InetProtocol getProtocol() {
+ return InetProtocol.TCP;
+ }
+
+ @Override
+ public int getReceiveBufferSize() throws IOException {
+ return this.socket.getReceiveBufferSize();
+ }
+
+ @Override
+ public void setReceiveBufferSize(@Nonnegative int receiveBufferSize)
throws IOException {
+ this.socket.setReceiveBufferSize(receiveBufferSize);
+ }
+
+ @Override
+ public boolean getReuseAddress() throws IOException {
+ return this.socket.getReuseAddress();
+ }
+
+ @Override
+ public void setReuseAddress(boolean reuseAddr) throws IOException {
+ this.socket.setReuseAddress(reuseAddr);
+ }
+
+ @Override
+ public ServerSocket getSocket() {
+ return this.socket;
+ }
+
+ @Override
+ public int getTimeout() throws IOException {
+ return this.socket.getSoTimeout();
+ }
+
+ @Override
+ public void setTimeout(@Nonnegative int timeout) throws IOException {
+ this.socket.setSoTimeout(timeout);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/TcpSocketAdapterImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,120 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.TcpSocketAdapter;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.net.SocketAddress;
+import javax.annotation.Nonnegative;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+
+public class TcpSocketAdapterImpl extends
AbstractClientSocketAdapter<Socket> implements TcpSocketAdapter {
+ public TcpSocketAdapterImpl(Socket socket) {
+ super(socket);
+ }
+
+ @Override
+ public void write(byte[] data) throws IOException {
+ this.write(data, this.socket.getRemoteSocketAddress());
+ }
+
+ @Override
+ public void write(byte[] data, SocketAddress remoteAddr) throws
IOException {
+ OutputStream outStream = this.socket.getOutputStream();
+
+ IOUtils.write(data, outStream);
+
+ outStream.flush();
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read() throws IOException {
+ return this.read(this.getProtocol().getDataSizeMax());
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read(@Nonnegative int bufferLen)
throws IOException {
+ return this.read(new byte[bufferLen]);
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read(byte[] buffer) throws
IOException {
+ int dataLen = IOUtils.read(this.socket.getInputStream(), buffer);
+
+ return new MutablePair<>(this.socket.getRemoteSocketAddress(),
ArrayUtils.subarray(buffer, 0, dataLen));
+ }
+
+ @Override
+ public void bind(SocketAddress bindSocketAddr) throws IOException {
+ this.socket.bind(bindSocketAddr);
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.socket.close();
+ }
+
+ @Override
+ public boolean isBound() {
+ return this.socket.isBound();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return this.socket.isClosed();
+ }
+
+ @Override
+ public InetProtocol getProtocol() {
+ return InetProtocol.TCP;
+ }
+
+ @Override
+ public int getReceiveBufferSize() throws IOException {
+ return this.socket.getReceiveBufferSize();
+ }
+
+ @Override
+ public void setReceiveBufferSize(@Nonnegative int receiveBufferSize)
throws IOException {
+ this.socket.setReceiveBufferSize(receiveBufferSize);
+ }
+
+ @Override
+ public boolean getReuseAddress() throws IOException {
+ return this.socket.getReuseAddress();
+ }
+
+ @Override
+ public void setReuseAddress(boolean reuseAddr) throws IOException {
+ this.socket.setReuseAddress(reuseAddr);
+ }
+
+ @Override
+ public int getSendBufferSize() throws IOException {
+ return this.socket.getSendBufferSize();
+ }
+
+ @Override
+ public void setSendBufferSize(@Nonnegative int sendBufferSize) throws
IOException {
+ this.socket.setSendBufferSize(sendBufferSize);
+ }
+
+ @Override
+ public Socket getSocket() {
+ return this.socket;
+ }
+
+ @Override
+ public int getTimeout() throws IOException {
+ return this.socket.getSoTimeout();
+ }
+
+ @Override
+ public void setTimeout(@Nonnegative int timeout) throws IOException {
+ this.socket.setSoTimeout(timeout);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/impl/UdpSocketAdapterImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,112 @@
+package gov.hhs.onc.dcdt.net.sockets.impl;
+
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.UdpSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.utils.ToolSocketUtils;
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.DatagramSocket;
+import java.net.SocketAddress;
+import javax.annotation.Nonnegative;
+import org.apache.commons.lang3.tuple.MutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+
+public class UdpSocketAdapterImpl extends
AbstractClientSocketAdapter<DatagramSocket> implements UdpSocketAdapter {
+ public UdpSocketAdapterImpl(DatagramSocket socket) {
+ super(socket);
+ }
+
+ @Override
+ public void write(byte[] data, SocketAddress remoteAddr) throws
IOException {
+ this.socket.send(ToolSocketUtils.createPacket(data, remoteAddr));
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read() throws IOException {
+ return this.read(this.getProtocol().getDataSizeMax());
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read(@Nonnegative int bufferLen)
throws IOException {
+ return this.read(new byte[bufferLen]);
+ }
+
+ @Override
+ public Pair<SocketAddress, byte[]> read(byte[] buffer) throws
IOException {
+ DatagramPacket packet = ToolSocketUtils.createPacket(buffer);
+
+ this.socket.receive(packet);
+
+ return new MutablePair<>(packet.getSocketAddress(),
packet.getData());
+ }
+
+ @Override
+ public void bind(SocketAddress bindSocketAddr) throws IOException {
+ this.socket.bind(bindSocketAddr);
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.socket.close();
+ }
+
+ @Override
+ public boolean isBound() {
+ return this.socket.isBound();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return this.socket.isClosed();
+ }
+
+ @Override
+ public InetProtocol getProtocol() {
+ return InetProtocol.UDP;
+ }
+
+ @Override
+ public int getReceiveBufferSize() throws IOException {
+ return this.socket.getReceiveBufferSize();
+ }
+
+ @Override
+ public void setReceiveBufferSize(@Nonnegative int receiveBufferSize)
throws IOException {
+ this.socket.setReceiveBufferSize(receiveBufferSize);
+ }
+
+ @Override
+ public boolean getReuseAddress() throws IOException {
+ return this.socket.getReuseAddress();
+ }
+
+ @Override
+ public void setReuseAddress(boolean reuseAddr) throws IOException {
+ this.socket.setReuseAddress(reuseAddr);
+ }
+
+ @Override
+ public int getSendBufferSize() throws IOException {
+ return this.socket.getSendBufferSize();
+ }
+
+ @Override
+ public void setSendBufferSize(@Nonnegative int sendBufferSize) throws
IOException {
+ this.socket.setSendBufferSize(sendBufferSize);
+ }
+
+ @Override
+ public DatagramSocket getSocket() {
+ return this.socket;
+ }
+
+ @Override
+ public int getTimeout() throws IOException {
+ return this.socket.getSoTimeout();
+ }
+
+ @Override
+ public void setTimeout(@Nonnegative int timeout) throws IOException {
+ this.socket.setSoTimeout(timeout);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/sockets/utils/ToolSocketUtils.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,19 @@
+package gov.hhs.onc.dcdt.net.sockets.utils;
+
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.SocketAddress;
+
+public abstract class ToolSocketUtils {
+ public static DatagramPacket createPacket(int bufferLen) throws
IOException {
+ return new DatagramPacket(new byte[bufferLen], bufferLen);
+ }
+
+ public static DatagramPacket createPacket(byte[] buffer) throws
IOException {
+ return new DatagramPacket(buffer, buffer.length);
+ }
+
+ public static DatagramPacket createPacket(byte[] buffer, SocketAddress
socketAddr) throws IOException {
+ return new DatagramPacket(buffer, buffer.length, socketAddr);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequest.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,13 @@
+package gov.hhs.onc.dcdt.service.dns.server;
+
+import gov.hhs.onc.dcdt.net.sockets.SocketRequest;
+import javax.annotation.Nonnegative;
+
+public interface DnsServerRequest extends SocketRequest {
+ public boolean hasQuerySize();
+
+ public int getQuerySize();
+
+ @Nonnegative
+ public int setQuerySize(@Nonnegative int querySize);
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequestProcessingException.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,49 @@
+package gov.hhs.onc.dcdt.service.dns.server;
+
+import gov.hhs.onc.dcdt.service.dns.DnsServiceException;
+import javax.annotation.Nullable;
+import org.xbill.DNS.Message;
+
+public class DnsServerRequestProcessingException extends
DnsServiceException {
+ private final static long serialVersionUID = 0L;
+
+ private Message reqMsg;
+ private Message respMsg;
+
+ public DnsServerRequestProcessingException(@Nullable Message reqMsg,
@Nullable Message respMsg) {
+ this(reqMsg, respMsg, null, null);
+ }
+
+ public DnsServerRequestProcessingException(@Nullable Message reqMsg,
@Nullable Message respMsg, String msg) {
+ this(reqMsg, respMsg, msg, null);
+ }
+
+ public DnsServerRequestProcessingException(@Nullable Message reqMsg,
@Nullable Message respMsg, Throwable cause) {
+ this(reqMsg, respMsg, null, cause);
+ }
+
+ public DnsServerRequestProcessingException(@Nullable Message reqMsg,
@Nullable Message respMsg, String msg, Throwable cause) {
+ super(msg, cause);
+
+ this.reqMsg = reqMsg;
+ this.respMsg = respMsg;
+ }
+
+ public boolean hasRequestMessage() {
+ return this.reqMsg != null;
+ }
+
+ @Nullable
+ public Message getRequestMessage() {
+ return this.reqMsg;
+ }
+
+ public boolean hasResponseMessage() {
+ return this.respMsg != null;
+ }
+
+ @Nullable
+ public Message getResponseMessage() {
+ return this.respMsg;
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerRequestProcessor.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,6 @@
+package gov.hhs.onc.dcdt.service.dns.server;
+
+import gov.hhs.onc.dcdt.net.sockets.SocketRequestProcessor;
+
+public interface DnsServerRequestProcessor extends
SocketRequestProcessor<DnsServerRequest> {
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerTcpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,6 @@
+package gov.hhs.onc.dcdt.service.dns.server;
+
+import gov.hhs.onc.dcdt.net.sockets.TcpSocketListener;
+
+public interface DnsServerTcpSocketListener extends
TcpSocketListener<DnsServerRequest, DnsServerRequestProcessor> {
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/DnsServerUdpSocketListener.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,6 @@
+package gov.hhs.onc.dcdt.service.dns.server;
+
+import gov.hhs.onc.dcdt.net.sockets.UdpSocketListener;
+
+public interface DnsServerUdpSocketListener extends
UdpSocketListener<DnsServerRequest, DnsServerRequestProcessor> {
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerRequestImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,38 @@
+package gov.hhs.onc.dcdt.service.dns.server.impl;
+
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.impl.AbstractSocketRequest;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequest;
+import gov.hhs.onc.dcdt.utils.ToolNumberUtils;
+import javax.annotation.Nonnegative;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+import org.xbill.DNS.Message;
+
+@Component("dnsServerReq")
+@Lazy
+@Scope("prototype")
+public class DnsServerRequestImpl extends AbstractSocketRequest implements
DnsServerRequest {
+ private int querySize = -1;
+
+ public DnsServerRequestImpl(InetProtocol protocol) {
+ super(protocol, Message.MAXLENGTH, Message.MAXLENGTH);
+ }
+
+ @Override
+ public boolean hasQuerySize() {
+ return ToolNumberUtils.isPositive(this.querySize);
+ }
+
+ @Override
+ public int getQuerySize() {
+ return this.querySize;
+ }
+
+ @Nonnegative
+ @Override
+ public int setQuerySize(@Nonnegative int querySize) {
+ return (this.querySize = querySize);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerRequestProcessorImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,100 @@
+package gov.hhs.onc.dcdt.service.dns.server.impl;
+
+import gov.hhs.onc.dcdt.config.instance.InstanceDnsConfig;
+import gov.hhs.onc.dcdt.dns.DnsException;
+import gov.hhs.onc.dcdt.dns.DnsMessageRcode;
+import gov.hhs.onc.dcdt.dns.DnsRecordType;
+import gov.hhs.onc.dcdt.dns.utils.ToolDnsMessageUtils;
+import gov.hhs.onc.dcdt.dns.utils.ToolDnsRecordUtils;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.impl.AbstractSocketRequestProcessor;
+import gov.hhs.onc.dcdt.service.dns.config.DnsServerConfig;
+import
gov.hhs.onc.dcdt.service.dns.server.DnsServerRequestProcessingException;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequest;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequestProcessor;
+import gov.hhs.onc.dcdt.utils.ToolClassUtils;
+import org.apache.commons.lang3.ArrayUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+import org.xbill.DNS.Message;
+import org.xbill.DNS.Name;
+import org.xbill.DNS.Record;
+
+@Component("dnsServerReqProcImpl")
+@Lazy
+@Scope("prototype")
+public class DnsServerRequestProcessorImpl extends
AbstractSocketRequestProcessor<DnsServerRequest> implements
DnsServerRequestProcessor {
+ private final static Logger LOGGER =
LoggerFactory.getLogger(DnsServerRequestProcessorImpl.class);
+
+ private DnsServerConfig serverConfig;
+
+ public DnsServerRequestProcessorImpl(DnsServerConfig serverConfig,
DnsServerRequest req) {
+ super(req);
+
+ this.serverConfig = serverConfig;
+ }
+
+ @Override
+ protected byte[] processError(byte[] reqData, Exception exception) {
+ return
ToolDnsMessageUtils.createErrorResponse(((DnsServerRequestProcessingException)
exception).getRequestMessage(), DnsMessageRcode.SERVFAIL).toWire();
+ }
+
+ @Override
+ protected byte[] processRequestInternal(byte[] reqData) throws
Exception {
+ InetProtocol protocol = this.req.getProtocol();
+ Message reqMsg = null, respMsg = null;
+ byte[] respData = null;
+
+ try {
+ respData = ToolDnsMessageUtils.toWire(protocol, (respMsg =
this.resolveQuery((reqMsg = ToolDnsMessageUtils.fromWire(protocol,
reqData)))));
+
+ LOGGER.trace(String.format("Resolved (class=%s) DNS server
query (protocol=%s, reqDataSize=%d, reqMsg={%s}): respDataSize=%d,
respMsg={%s}",
+ ToolClassUtils.getName(this), protocol.name(),
reqData.length, reqMsg, ArrayUtils.getLength(respData), respMsg));
+
+ return respData;
+ } catch (Exception e) {
+ throw new DnsServerRequestProcessingException(reqMsg, respMsg,
String.format(
+ "Unable to resolve (class=%s) DNS server query
(protocol=%s, reqDataSize=%d, reqMsg={%s}): respDataSize=%d, respMsg={%s}",
+ ToolClassUtils.getName(this), protocol.name(),
reqData.length, reqMsg, ArrayUtils.getLength(respData), respMsg), e);
+ }
+ }
+
+ private Message resolveQuery(Message reqMsg) throws DnsException {
+ Message respMsg = ToolDnsMessageUtils.createResponse(reqMsg);
+ Record questionRecord = reqMsg.getQuestion();
+
+ if (questionRecord == null) {
+ ToolDnsMessageUtils.setRcode(respMsg, DnsMessageRcode.FORMERR);
+
+ return respMsg;
+ }
+
+ DnsRecordType questionRecordType =
ToolDnsRecordUtils.findByType(questionRecord.getType());
+ Name questionName;
+
+ if (questionRecordType == null) {
+ ToolDnsMessageUtils.setRcode(respMsg, DnsMessageRcode.NXRRSET);
+
+ return respMsg;
+ } else if (!(questionName = questionRecord.getName()).isAbsolute()
|| questionName.isWild()) {
+ ToolDnsMessageUtils.setRcode(respMsg, DnsMessageRcode.REFUSED);
+
+ return respMsg;
+ }
+
+ InstanceDnsConfig authoritativeDnsConfig =
this.serverConfig.findAuthoritativeDnsConfig(questionRecord);
+
+ if (authoritativeDnsConfig != null) {
+ ToolDnsMessageUtils.setAnswers(respMsg,
authoritativeDnsConfig.findAnswers(questionRecord));
+ // noinspection ConstantConditions
+ ToolDnsMessageUtils.setAuthorities(respMsg, true,
authoritativeDnsConfig.getSoaRecordConfig().toRecord());
+ } else {
+ ToolDnsMessageUtils.setRcode(respMsg, DnsMessageRcode.REFUSED);
+ }
+
+ return respMsg;
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerTcpSocketListenerImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,68 @@
+package gov.hhs.onc.dcdt.service.dns.server.impl;
+
+import gov.hhs.onc.dcdt.beans.Phase;
+import gov.hhs.onc.dcdt.context.AutoStartup;
+import gov.hhs.onc.dcdt.dns.utils.ToolDnsMessageUtils;
+import gov.hhs.onc.dcdt.net.InetProtocol;
+import gov.hhs.onc.dcdt.net.sockets.TcpServerSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.TcpSocketAdapter;
+import gov.hhs.onc.dcdt.net.sockets.impl.AbstractTcpSocketListener;
+import gov.hhs.onc.dcdt.service.dns.config.DnsServerConfig;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequest;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequestProcessor;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerTcpSocketListener;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import javax.annotation.Resource;
+import org.apache.commons.lang3.ArrayUtils;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.context.annotation.Scope;
+import org.springframework.core.task.AsyncListenableTaskExecutor;
+import org.springframework.stereotype.Component;
+
+@AutoStartup(false)
+@Component("dnsServerSocketListenerTcp")
+@Lazy
+@Phase(Phase.PHASE_PRECEDENCE_HIGHEST)
+@Scope("prototype")
+public class DnsServerTcpSocketListenerImpl extends
AbstractTcpSocketListener<DnsServerRequest, DnsServerRequestProcessor>
implements
+ DnsServerTcpSocketListener {
+ private DnsServerConfig serverConfig;
+
+ public DnsServerTcpSocketListenerImpl(DnsServerConfig serverConfig) {
+ super(DnsServerRequest.class, DnsServerRequestProcessor.class,
serverConfig.toSocketAddress());
+
+ this.serverConfig = serverConfig;
+ }
+
+ @Override
+ protected DnsServerRequestProcessor createRequestProcessor(Object ...
reqProcArgs) {
+ return super.createRequestProcessor(ArrayUtils.add(reqProcArgs, 0,
this.serverConfig));
+ }
+
+ @Override
+ protected TcpSocketAdapter readRequest(TcpServerSocketAdapter
listenSocketAdapter, DnsServerRequest req) throws IOException {
+ TcpSocketAdapter reqSocketAdapter =
super.readRequest(listenSocketAdapter, req);
+ ByteBuffer reqBuffer = req.getRequestBuffer();
+ int reqQuerySize =
req.setQuerySize(ToolDnsMessageUtils.parseQuerySizeData(reqBuffer.get(0),
reqBuffer.get(1)));
+
+ reqBuffer.limit(reqBuffer.limit() + reqQuerySize);
+ reqBuffer.put(reqSocketAdapter.read(reqQuerySize).getRight());
+
+ return reqSocketAdapter;
+ }
+
+ @Override
+ protected DnsServerRequest createRequest(InetProtocol protocol) {
+ DnsServerRequest req = super.createRequest(protocol);
+
req.getRequestBuffer().limit(ToolDnsMessageUtils.DATA_SIZE_DNS_MSG_QUERY_SIZE_PREFIX);
+
+ return req;
+ }
+
+ @Override
+ @Resource(name = "taskExecServiceDnsServerReqQuery")
+ protected void setTaskExecutor(AsyncListenableTaskExecutor taskExec) {
+ super.setTaskExecutor(taskExec);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/main/java/gov/hhs/onc/dcdt/service/dns/server/impl/DnsServerUdpSocketListenerImpl.java
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,42 @@
+package gov.hhs.onc.dcdt.service.dns.server.impl;
+
+import gov.hhs.onc.dcdt.beans.Phase;
+import gov.hhs.onc.dcdt.context.AutoStartup;
+import gov.hhs.onc.dcdt.net.sockets.impl.AbstractUdpSocketListener;
+import gov.hhs.onc.dcdt.service.dns.config.DnsServerConfig;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequest;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerRequestProcessor;
+import gov.hhs.onc.dcdt.service.dns.server.DnsServerUdpSocketListener;
+import javax.annotation.Resource;
+import org.apache.commons.lang3.ArrayUtils;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.context.annotation.Scope;
+import org.springframework.core.task.AsyncListenableTaskExecutor;
+import org.springframework.stereotype.Component;
+
+@AutoStartup(false)
+@Component("dnsServerSocketListenerUdp")
+@Lazy
+@Phase(Phase.PHASE_PRECEDENCE_HIGHEST)
+@Scope("prototype")
+public class DnsServerUdpSocketListenerImpl extends
AbstractUdpSocketListener<DnsServerRequest, DnsServerRequestProcessor>
implements
+ DnsServerUdpSocketListener {
+ private DnsServerConfig serverConfig;
+
+ public DnsServerUdpSocketListenerImpl(DnsServerConfig serverConfig) {
+ super(DnsServerRequest.class, DnsServerRequestProcessor.class,
serverConfig.toSocketAddress());
+
+ this.serverConfig = serverConfig;
+ }
+
+ @Override
+ protected DnsServerRequestProcessor createRequestProcessor(Object ...
reqProcArgs) {
+ return super.createRequestProcessor(ArrayUtils.add(reqProcArgs, 0,
this.serverConfig));
+ }
+
+ @Override
+ @Resource(name = "taskExecServiceDnsServerReqQuery")
+ protected void setTaskExecutor(AsyncListenableTaskExecutor taskExec) {
+ super.setTaskExecutor(taskExec);
+ }
+}
=======================================
--- /dev/null
+++
/dcdt-service-dns/src/test/resources/META-INF/service/dns/service-dns-test.properties
Thu May 22 06:39:17 2014 UTC
@@ -0,0 +1,4 @@
+#====================================================================================================
+# DNS SERVICE
+#====================================================================================================
+dcdt.test.func.service.dns.lookup.concurrent.threads=50
=======================================
--- /dcdt-core/src/main/java/gov/hhs/onc/dcdt/net/ToolSocketOption.java Sat
Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,23 +0,0 @@
-package gov.hhs.onc.dcdt.net;
-
-import java.net.SocketOption;
-
-public class ToolSocketOption<T> implements SocketOption<T> {
- private String name;
- private Class<T> type;
-
- public ToolSocketOption(String name, Class<T> type) {
- this.name = name;
- this.type = type;
- }
-
- @Override
- public String name() {
- return this.name;
- }
-
- @Override
- public Class<T> type() {
- return this.type;
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListener.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,19 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import gov.hhs.onc.dcdt.beans.ToolBean;
-import java.io.IOException;
-import java.nio.channels.ByteChannel;
-import java.nio.channels.NetworkChannel;
-import java.nio.channels.SelectableChannel;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
-import javax.annotation.Nullable;
-import org.springframework.context.ApplicationContextAware;
-
-public interface ChannelListener<T extends SelectableChannel &
NetworkChannel, U extends SelectableChannel & ByteChannel & NetworkChannel,
V extends SelectableChannel & ByteChannel & NetworkChannel, W extends
SelectionAttachment, X extends ChannelListenerDataProcessor, Y extends
ChannelListenerDataProcessorCallback<X, W>>
- extends ApplicationContextAware, ToolBean {
- @Nullable
- public SelectionOperationType processSelection(SelectionKey selKey)
throws IOException;
-
- public void register(Selector selector) throws IOException;
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerDataProcessor.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,7 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import gov.hhs.onc.dcdt.beans.ToolBean;
-import java.util.concurrent.Callable;
-
-public interface ChannelListenerDataProcessor extends Callable<byte[]>,
ToolBean {
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerDataProcessorCallback.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,7 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import org.springframework.util.concurrent.ListenableFutureCallback;
-
-public interface ChannelListenerDataProcessorCallback<T extends
ChannelListenerDataProcessor, U extends SelectionAttachment> extends
- ListenableFutureCallback<byte[]> {
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelListenerSelector.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,13 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import gov.hhs.onc.dcdt.beans.ToolLifecycleBean;
-import java.util.List;
-import javax.annotation.Nullable;
-
-public interface ChannelListenerSelector extends ToolLifecycleBean {
- public List<ChannelListener<?, ?, ?, ?, ?, ?>> getChannelListeners();
-
- public void setChannelListeners(@Nullable
ChannelListener<?, ?, ?, ?, ?, ?> ... channelListeners);
-
- public void setChannelListeners(@Nullable
Iterable<ChannelListener<?, ?, ?, ?, ?, ?>> channelListeners);
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/ChannelSocketOptions.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,8 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import gov.hhs.onc.dcdt.net.ToolSocketOption;
-import java.net.SocketOption;
-
-public interface ChannelSocketOptions {
- public final static SocketOption<Boolean> CHANNEL_BLOCKING = new
ToolSocketOption<>("CHANNEL_BLOCKING", Boolean.class);
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/SelectionAttachment.java
Tue Feb 25 21:16:43 2014 UTC
+++ /dev/null
@@ -1,26 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import java.net.SocketAddress;
-import java.nio.ByteBuffer;
-import javax.annotation.Nullable;
-
-public interface SelectionAttachment {
- public InetProtocol getProtocol();
-
- public void setProtocol(InetProtocol protocol);
-
- public ByteBuffer getRequestBuffer();
-
- public void setRequestBuffer(ByteBuffer reqBuffer);
-
- public ByteBuffer getResponseBuffer();
-
- public void setResponseBuffer(ByteBuffer respBuffer);
-
- @Nullable
- public SocketAddress getSocketAddress();
-
- @Nullable
- public SocketAddress setSocketAddress(@Nullable SocketAddress
socketAddr);
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/SelectionOperationType.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,25 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels;
-
-import java.nio.channels.SelectionKey;
-
-public enum SelectionOperationType {
- READ(SelectionKey.OP_READ), WRITE(SelectionKey.OP_WRITE),
ACCEPT(SelectionKey.OP_ACCEPT), CONNECT(SelectionKey.OP_CONNECT);
-
- private final int op;
-
- private SelectionOperationType(int op) {
- this.op = op;
- }
-
- public boolean isReady(SelectionKey selKey) {
- return this.isReady(selKey.readyOps());
- }
-
- public boolean isReady(int readyOps) {
- return (readyOps & this.op) != 0;
- }
-
- public int getOperation() {
- return this.op;
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListener.java
Tue Feb 25 21:16:43 2014 UTC
+++ /dev/null
@@ -1,208 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.beans.impl.AbstractToolBean;
-import gov.hhs.onc.dcdt.beans.utils.ToolBeanFactoryUtils;
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListener;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessor;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessorCallback;
-import gov.hhs.onc.dcdt.nio.channels.ChannelSocketOptions;
-import gov.hhs.onc.dcdt.nio.channels.SelectionAttachment;
-import gov.hhs.onc.dcdt.nio.channels.SelectionOperationType;
-import gov.hhs.onc.dcdt.nio.utils.ToolBufferUtils;
-import gov.hhs.onc.dcdt.nio.utils.ToolChannelUtils;
-import gov.hhs.onc.dcdt.utils.ToolArrayUtils;
-import gov.hhs.onc.dcdt.utils.ToolClassUtils;
-import gov.hhs.onc.dcdt.utils.ToolMapUtils;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.SocketOption;
-import java.net.StandardSocketOptions;
-import java.nio.ByteBuffer;
-import java.nio.channels.ByteChannel;
-import java.nio.channels.ClosedChannelException;
-import java.nio.channels.NetworkChannel;
-import java.nio.channels.SelectableChannel;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
-import java.util.List;
-import java.util.Map;
-import javax.annotation.Nullable;
-import org.apache.commons.collections4.map.LinkedMap;
-import org.apache.commons.lang3.tuple.MutablePair;
-import org.apache.commons.lang3.tuple.Pair;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.BeansException;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.support.AbstractApplicationContext;
-import org.springframework.core.task.AsyncListenableTaskExecutor;
-import org.springframework.util.concurrent.ListenableFutureCallback;
-import org.springframework.util.concurrent.ListenableFutureTask;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractChannelListener<T extends SelectableChannel
& NetworkChannel, U extends SelectableChannel & ByteChannel &
NetworkChannel, V extends SelectableChannel & ByteChannel & NetworkChannel,
W extends SelectionAttachment, X extends ChannelListenerDataProcessor, Y
extends ChannelListenerDataProcessorCallback<X, W>>
- extends AbstractToolBean implements ChannelListener<T, U, V, W, X, Y> {
- @SuppressWarnings({ "unchecked" })
- protected final static List<? extends Pair<? extends
SocketOption<?>, ?>> OPT_PAIRS_CHANNEL = ToolArrayUtils.asList(new
MutablePair<>(
- StandardSocketOptions.SO_REUSEADDR, true), new
MutablePair<>(ChannelSocketOptions.CHANNEL_BLOCKING, false));
-
- protected AbstractApplicationContext appContext;
- protected AsyncListenableTaskExecutor taskExec;
- protected Class<T> listenChannelClass;
- protected Class<U> readChannelClass;
- protected Class<V> writeChannelClass;
- protected Class<W> attachmentClass;
- protected Class<X> dataProcClass;
- protected Class<Y> dataProcCallbackClass;
- protected SelectionOperationType listenOpType;
- protected SelectionOperationType readOpType;
- protected SelectionOperationType writeOpType;
- protected InetProtocol protocol;
- protected InetSocketAddress bindSocketAddr;
-
- private final static Logger LOGGER =
LoggerFactory.getLogger(AbstractChannelListener.class);
-
- protected AbstractChannelListener(Class<T> listenChannelClass,
Class<U> readChannelClass, Class<V> writeChannelClass, Class<W>
attachmentClass,
- Class<X> dataProcClass, Class<Y> dataProcCallbackClass,
SelectionOperationType listenOpType, SelectionOperationType readOpType,
- SelectionOperationType writeOpType, InetProtocol protocol,
InetSocketAddress bindSocketAddr) {
- this.listenChannelClass = listenChannelClass;
- this.readChannelClass = readChannelClass;
- this.writeChannelClass = writeChannelClass;
- this.attachmentClass = attachmentClass;
- this.dataProcClass = dataProcClass;
- this.dataProcCallbackClass = dataProcCallbackClass;
- this.listenOpType = listenOpType;
- this.readOpType = readOpType;
- this.writeOpType = writeOpType;
- this.protocol = protocol;
- this.bindSocketAddr = bindSocketAddr;
- }
-
- @Nullable
- @Override
- public SelectionOperationType processSelection(SelectionKey selKey)
throws IOException {
- Pair<SelectionOperationType, ? extends Class<? extends
SelectableChannel>> selOp = this.getSelectedOperation(selKey);
-
- if (selOp != null) {
- SelectionOperationType selOpType = selOp.getLeft();
-
- W selAttachment =
this.attachmentClass.cast(selKey.attachment());
- boolean selProcessed;
-
- if (selOpType == this.readOpType) {
- selProcessed = this.processReadOperation(selKey,
this.readChannelClass.cast(selKey.channel()), selAttachment);
- } else if (selOpType == this.writeOpType) {
- selProcessed = this.processWriteOperation(selKey,
this.writeChannelClass.cast(selKey.channel()), selAttachment);
- } else {
- selProcessed = this.processListenOperation(selKey,
this.listenChannelClass.cast(selKey.channel()), selAttachment);
- }
-
- if (selProcessed) {
- LOGGER.trace(String.format("Channel listener (class=%s)
processed selection (opType=%s, channelClass=%s, attachmentClass=%s).",
- ToolClassUtils.getName(this), selOpType.name(),
ToolClassUtils.getName(selOp.getRight()),
ToolClassUtils.getName(selAttachment)));
- }
-
- return selOpType;
- } else {
- return null;
- }
- }
-
- @Override
- public void register(Selector selector) throws IOException {
- try {
- this.createListenChannel().register(selector,
this.listenOpType.getOperation(), this.createAttachment());
- } catch (ClosedChannelException ignored) {
- }
- }
-
- protected static boolean isOperationSelected(SelectionKey selKey,
SelectionOperationType selOpType, Class<? extends SelectableChannel>
selOpChannelClass) {
- return selOpType.isReady(selKey) &&
ToolClassUtils.isAssignable(selKey.channel().getClass(), selOpChannelClass);
- }
-
- protected boolean processWriteOperation(SelectionKey selKey, V
writeChannel, W selAttachment) throws IOException {
- selKey.attach(this.createAttachment());
- selKey.interestOps(this.readOpType.getOperation());
- selKey.selector().wakeup();
-
- return true;
- }
-
- protected boolean processReadOperation(SelectionKey selKey, U
readChannel, W selAttachment) throws IOException {
-
this.executeDataProcessorTask(ToolBufferUtils.flip(selAttachment.getRequestBuffer()),
this.createDataProcessorCallbacks(selKey, selAttachment));
-
- return true;
- }
-
- protected boolean processListenOperation(SelectionKey selKey, T
listenChannel, W selAttachment) throws IOException {
- selKey.selector().wakeup();
-
- return true;
- }
-
- protected void executeDataProcessorTask(ByteBuffer reqBuffer,
Iterable<? extends ListenableFutureCallback<byte[]>> dataProcCallbacks) {
- ListenableFutureTask<byte[]> dataProcFutureTask = new
ListenableFutureTask<>(this.createDataProcessor(this.protocol,
ToolBufferUtils.get(reqBuffer)));
-
- for (ListenableFutureCallback<byte[]> dataProcCallback :
dataProcCallbacks) {
- dataProcFutureTask.addCallback(dataProcCallback);
- }
-
- this.taskExec.execute(dataProcFutureTask);
- }
-
- @Nullable
- protected Pair<SelectionOperationType, ? extends Class<? extends
SelectableChannel>> getSelectedOperation(SelectionKey selKey) {
- SelectionOperationType selOpType;
- Class<? extends SelectableChannel> selChannelClass;
-
- return (isOperationSelected(selKey, (selOpType = this.readOpType),
(selChannelClass = this.readChannelClass))
- || isOperationSelected(selKey, (selOpType = this.writeOpType),
(selChannelClass = this.writeChannelClass)) || isOperationSelected(selKey,
- (selOpType = this.listenOpType), (selChannelClass =
this.listenChannelClass))) ? new MutablePair<>(selOpType,
selChannelClass) : null;
- }
-
- protected List<? extends Y> createDataProcessorCallbacks(SelectionKey
selKey, W selAttachment) {
- return ToolBeanFactoryUtils.createBeansOfType(this.appContext,
this.dataProcCallbackClass, this.readOpType, this.writeOpType,
this.protocol, selKey,
- selAttachment);
- }
-
- protected X createDataProcessor(Object ... beanCreationArgs) {
- return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.dataProcClass, beanCreationArgs);
- }
-
- protected W createAttachment() {
- return ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.attachmentClass, this.protocol);
- }
-
- protected T createListenChannel() throws IOException {
- return
this.listenChannelClass.cast(this.initializeChannel(ToolBeanFactoryUtils.createBeanOfType(this.appContext,
this.listenChannelClass)).bind(
- this.bindSocketAddr));
- }
-
- protected <Z extends SelectableChannel & NetworkChannel> Z
initializeChannel(Z channel) throws IOException {
- Map<SocketOption<?>, Object> optMap = ToolMapUtils.putAll(new
LinkedMap<SocketOption<?>, Object>(), OPT_PAIRS_CHANNEL);
- Class<? extends SelectableChannel> channelClass =
channel.getClass();
- int bufferSize = this.protocol.getDataSizeMax();
-
- if (ToolClassUtils.isAssignable(channelClass,
this.readChannelClass)) {
- // noinspection ConstantConditions
- optMap.put(StandardSocketOptions.SO_RCVBUF, bufferSize);
- }
-
- if (ToolClassUtils.isAssignable(channelClass,
this.writeChannelClass)) {
- // noinspection ConstantConditions
- optMap.put(StandardSocketOptions.SO_SNDBUF, bufferSize);
- }
-
- return ToolChannelUtils.setOptions(channel, optMap);
- }
-
- @Override
- public void setApplicationContext(ApplicationContext appContext)
throws BeansException {
- this.appContext = (AbstractApplicationContext) appContext;
- }
-
- protected void setTaskExecutor(AsyncListenableTaskExecutor taskExec) {
- this.taskExec = taskExec;
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerDataProcessor.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,16 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.beans.impl.AbstractToolBean;
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessor;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractChannelListenerDataProcessor extends
AbstractToolBean implements ChannelListenerDataProcessor {
- protected InetProtocol protocol;
- protected byte[] reqData;
-
- protected AbstractChannelListenerDataProcessor(InetProtocol protocol,
byte[] reqData) {
- this.protocol = protocol;
- this.reqData = reqData;
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerDataProcessorCallback.java
Sat Feb 22 21:15:33 2014 UTC
+++ /dev/null
@@ -1,47 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessor;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessorCallback;
-import gov.hhs.onc.dcdt.nio.channels.SelectionAttachment;
-import gov.hhs.onc.dcdt.nio.channels.SelectionOperationType;
-import gov.hhs.onc.dcdt.nio.utils.ToolBufferUtils;
-import java.nio.channels.SelectionKey;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractChannelListenerDataProcessorCallback<T
extends ChannelListenerDataProcessor, U extends SelectionAttachment>
implements
- ChannelListenerDataProcessorCallback<T, U> {
- protected Class<T> dataProcClass;
- protected SelectionOperationType readOpType;
- protected SelectionOperationType writeOpType;
- protected InetProtocol protocol;
- protected SelectionKey selKey;
- protected U selAttachment;
-
- public AbstractChannelListenerDataProcessorCallback(Class<T>
dataProcClass, SelectionOperationType readOpType, SelectionOperationType
writeOpType,
- InetProtocol protocol, SelectionKey selKey, U selAttachment) {
- this.dataProcClass = dataProcClass;
- this.readOpType = readOpType;
- this.writeOpType = writeOpType;
- this.protocol = protocol;
- this.selKey = selKey;
- this.selAttachment = selAttachment;
- }
-
- @Override
- public void onSuccess(byte[] respData) {
- this.setResponseData(respData);
- }
-
- @Override
- public void onFailure(Throwable th) {
- }
-
- protected void setResponseData(byte[] respData) {
-
ToolBufferUtils.clear(this.selAttachment.getResponseBuffer()).put(respData);
-
- this.selKey.interestOps(this.writeOpType.getOperation());
-
- this.selKey.selector().wakeup();
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractChannelListenerSelector.java
Tue Feb 25 21:16:43 2014 UTC
+++ /dev/null
@@ -1,122 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.beans.impl.AbstractToolLifecycleBean;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListener;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerSelector;
-import gov.hhs.onc.dcdt.nio.channels.SelectionOperationType;
-import gov.hhs.onc.dcdt.utils.ToolArrayUtils;
-import gov.hhs.onc.dcdt.utils.ToolClassUtils;
-import gov.hhs.onc.dcdt.utils.ToolCollectionUtils;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.concurrent.Callable;
-import java.util.concurrent.FutureTask;
-import javax.annotation.Nullable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public abstract class AbstractChannelListenerSelector extends
AbstractToolLifecycleBean implements ChannelListenerSelector {
- protected class ChannelListenerSelectorDaemon implements
Callable<Void> {
- @Nullable
- @Override
- public Void call() throws Exception {
- Iterator<SelectionKey> selKeysIterator;
- SelectionKey selKey;
- SelectionOperationType selOpType;
-
- while (AbstractChannelListenerSelector.this.isRunning()) {
- if (AbstractChannelListenerSelector.this.selector.select()
== 0) {
- continue;
- }
-
- while ((selKeysIterator =
AbstractChannelListenerSelector.this.selector.selectedKeys().iterator()).hasNext()
- && ((selKey = selKeysIterator.next()) != null)) {
- selKeysIterator.remove();
-
- if (!selKey.isValid()) {
- continue;
- }
-
- selOpType = null;
-
- for (ChannelListener<?, ?, ?, ?, ?, ?>
channelListener : AbstractChannelListenerSelector.this.channelListeners) {
- if ((selOpType =
channelListener.processSelection(selKey)) != null) {
- break;
- }
- }
-
- if (selOpType == null) {
-
AbstractChannelListenerSelector.LOGGER.warn(String.format(
- "Channel listener selector (class=%s)
selection (readyOps=%d, channelClass=%s, attachmentClass=%s) not
processed.",
-
ToolClassUtils.getName(AbstractChannelListenerSelector.this),
selKey.readyOps(), ToolClassUtils.getName(selKey.channel()),
- ToolClassUtils.getName(selKey.attachment())));
-
- selKey.cancel();
- }
- }
- }
-
- return null;
- }
- }
-
- protected List<ChannelListener<?, ?, ?, ?, ?, ?>> channelListeners =
new ArrayList<>();
- protected Selector selector;
- protected FutureTask<Void> selectorDaemonTask;
-
- private final static Logger LOGGER =
LoggerFactory.getLogger(AbstractChannelListenerSelector.class);
-
- @Override
- public boolean isRunning() {
- return super.isRunning() && (this.selector != null) &&
this.selector.isOpen();
- }
-
- @Override
- protected void stopInternal() throws Exception {
- if ((this.selectorDaemonTask != null)
&& !this.selectorDaemonTask.isDone()) {
- this.selectorDaemonTask.cancel(true);
- }
-
- for (SelectionKey selKey : this.selector.keys()) {
- try {
- selKey.channel().close();
- } catch (Exception ignored) {
- }
- }
-
- try {
- this.selector.close();
- } catch (Exception ignored) {
- }
- }
-
- @Override
- protected void startInternal() throws Exception {
- this.selector = Selector.open();
-
- for (ChannelListener<?, ?, ?, ?, ?, ?> channelListener :
this.channelListeners) {
- channelListener.register(this.selector);
- }
-
- this.taskExec.execute((this.selectorDaemonTask = new
FutureTask<>(new ChannelListenerSelectorDaemon())));
- }
-
- @Override
- public List<ChannelListener<?, ?, ?, ?, ?, ?>> getChannelListeners() {
- return this.channelListeners;
- }
-
- @Override
- public void setChannelListeners(@Nullable
ChannelListener<?, ?, ?, ?, ?, ?> ... channelListeners) {
- this.setChannelListeners(ToolArrayUtils.asList(channelListeners));
- }
-
- @Override
- public void setChannelListeners(@Nullable
Iterable<ChannelListener<?, ?, ?, ?, ?, ?>> channelListeners) {
- this.channelListeners.clear();
- ToolCollectionUtils.addAll(this.channelListeners,
channelListeners);
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractSelectionAttachment.java
Tue Feb 25 21:16:43 2014 UTC
+++ /dev/null
@@ -1,64 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.SelectionAttachment;
-import java.net.SocketAddress;
-import java.nio.ByteBuffer;
-import javax.annotation.Nonnegative;
-import javax.annotation.Nullable;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractSelectionAttachment implements
SelectionAttachment {
- protected InetProtocol protocol;
- protected ByteBuffer reqBuffer;
- protected ByteBuffer respBuffer;
- protected SocketAddress socketAddr;
-
- protected AbstractSelectionAttachment(InetProtocol protocol,
@Nonnegative int reqBufferSize, @Nonnegative int respBufferSize) {
- this.protocol = protocol;
- this.reqBuffer = ByteBuffer.allocate(reqBufferSize);
- this.respBuffer = ByteBuffer.allocate(respBufferSize);
- }
-
- @Override
- public InetProtocol getProtocol() {
- return this.protocol;
- }
-
- @Override
- public void setProtocol(InetProtocol protocol) {
- this.protocol = protocol;
- }
-
- @Override
- public ByteBuffer getRequestBuffer() {
- return this.reqBuffer;
- }
-
- @Override
- public void setRequestBuffer(ByteBuffer reqBuffer) {
- this.reqBuffer = reqBuffer;
- }
-
- @Override
- public ByteBuffer getResponseBuffer() {
- return this.respBuffer;
- }
-
- @Override
- public void setResponseBuffer(ByteBuffer respBuffer) {
- this.respBuffer = respBuffer;
- }
-
- @Nullable
- @Override
- public SocketAddress getSocketAddress() {
- return this.socketAddr;
- }
-
- @Nullable
- @Override
- public SocketAddress setSocketAddress(@Nullable SocketAddress
socketAddr) {
- return (this.socketAddr = socketAddr);
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractTcpChannelListener.java
Mon May 5 03:47:29 2014 UTC
+++ /dev/null
@@ -1,76 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessor;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessorCallback;
-import gov.hhs.onc.dcdt.nio.channels.SelectionAttachment;
-import gov.hhs.onc.dcdt.nio.channels.SelectionOperationType;
-import gov.hhs.onc.dcdt.nio.utils.ToolBufferUtils;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.ServerSocketChannel;
-import java.nio.channels.SocketChannel;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractTcpChannelListener<T extends
SelectionAttachment, U extends ChannelListenerDataProcessor, V extends
ChannelListenerDataProcessorCallback<U, T>>
- extends AbstractChannelListener<ServerSocketChannel, SocketChannel,
SocketChannel, T, U, V> {
- protected AbstractTcpChannelListener(Class<T> attachmentClass,
Class<U> dataProcClass, Class<V> dataProcCallbackClass, InetSocketAddress
bindSocketAddr) {
- super(ServerSocketChannel.class, SocketChannel.class,
SocketChannel.class, attachmentClass, dataProcClass, dataProcCallbackClass,
- SelectionOperationType.ACCEPT, SelectionOperationType.READ,
SelectionOperationType.WRITE, InetProtocol.TCP, bindSocketAddr);
- }
-
- @Override
- protected boolean processWriteOperation(SelectionKey selKey,
SocketChannel writeChannel, T selAttachment) throws IOException {
- ByteBuffer respBuffer = selAttachment.getResponseBuffer();
-
- if (respBuffer.position() > 0) {
- writeChannel.write(ToolBufferUtils.flip(respBuffer));
-
- if (respBuffer.hasRemaining()) {
- respBuffer.compact();
-
- selKey.selector().wakeup();
-
- return false;
- }
- }
-
- return super.processWriteOperation(selKey, writeChannel,
selAttachment);
- }
-
- @Override
- protected boolean processReadOperation(SelectionKey selKey,
SocketChannel readChannel, T selAttachment) throws IOException {
- ByteBuffer reqBuffer = selAttachment.getRequestBuffer();
-
- if (readChannel.read(reqBuffer) == -1) {
- readChannel.close();
- selKey.cancel();
- } else if (!reqBuffer.hasRemaining()) {
- if (reqBuffer.position() > 0) {
- return super.processReadOperation(selKey, readChannel,
selAttachment);
- } else {
- selKey.cancel();
- }
- }
-
- selKey.selector().wakeup();
-
- return false;
- }
-
- @Override
- @SuppressWarnings({ "unchecked" })
- protected boolean processListenOperation(SelectionKey selKey,
ServerSocketChannel listenChannel, T selAttachment) throws IOException {
- SocketChannel readChannel = listenChannel.accept();
-
- if (readChannel != null) {
-
this.initializeChannel(readChannel).register(selKey.selector(),
this.readOpType.getOperation(), this.createAttachment());
-
- return super.processListenOperation(selKey, listenChannel,
selAttachment);
- } else {
- return false;
- }
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/channels/impl/AbstractUdpChannelListener.java
Sat Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,34 +0,0 @@
-package gov.hhs.onc.dcdt.nio.channels.impl;
-
-import gov.hhs.onc.dcdt.net.InetProtocol;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessor;
-import gov.hhs.onc.dcdt.nio.channels.ChannelListenerDataProcessorCallback;
-import gov.hhs.onc.dcdt.nio.channels.SelectionAttachment;
-import gov.hhs.onc.dcdt.nio.channels.SelectionOperationType;
-import gov.hhs.onc.dcdt.nio.utils.ToolBufferUtils;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.nio.channels.DatagramChannel;
-import java.nio.channels.SelectionKey;
-
-@SuppressWarnings({ "SpringJavaAutowiringInspection" })
-public abstract class AbstractUdpChannelListener<T extends
SelectionAttachment, U extends ChannelListenerDataProcessor, V extends
ChannelListenerDataProcessorCallback<U, T>>
- extends AbstractChannelListener<DatagramChannel, DatagramChannel,
DatagramChannel, T, U, V> {
- protected AbstractUdpChannelListener(Class<T> attachmentClass,
Class<U> dataProcClass, Class<V> dataProcCallbackClass, InetSocketAddress
bindSocketAddr) {
- super(DatagramChannel.class, DatagramChannel.class,
DatagramChannel.class, attachmentClass, dataProcClass,
dataProcCallbackClass,
- SelectionOperationType.READ, SelectionOperationType.READ,
SelectionOperationType.WRITE, InetProtocol.UDP, bindSocketAddr);
- }
-
- @Override
- protected boolean processWriteOperation(SelectionKey selKey,
DatagramChannel writeChannel, T selAttachment) throws IOException {
-
writeChannel.send(ToolBufferUtils.flip(selAttachment.getResponseBuffer()),
selAttachment.getSocketAddress());
-
- return super.processWriteOperation(selKey, writeChannel,
selAttachment);
- }
-
- @Override
- protected boolean processReadOperation(SelectionKey selKey,
DatagramChannel readChannel, T selAttachment) throws IOException {
- return
(selAttachment.setSocketAddress(readChannel.receive(selAttachment.getRequestBuffer())) !=
null)
- && (selAttachment.getRequestBuffer().position() > 0) &&
super.processReadOperation(selKey, readChannel, selAttachment);
- }
-}
=======================================
---
/dcdt-core/src/main/java/gov/hhs/onc/dcdt/nio/utils/ToolChannelUtils.java
Sat Feb 22 21:15:33 2014 UTC
+++ /dev/null
@@ -1,35 +0,0 @@
-package gov.hhs.onc.dcdt.nio.utils;
-
-import gov.hhs.onc.dcdt.net.ToolSocketOption;
-import gov.hhs.onc.dcdt.nio.channels.ChannelSocketOptions;
-import gov.hhs.onc.dcdt.utils.ToolClassUtils;
-import gov.hhs.onc.dcdt.utils.ToolMapUtils;
-import java.io.IOException;
-import java.net.SocketOption;
-import java.nio.channels.NetworkChannel;
-import java.nio.channels.SelectableChannel;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Map.Entry;
-
-public abstract class ToolChannelUtils {
- @SuppressWarnings({ "unchecked" })
- public static <T extends SelectableChannel & NetworkChannel> T
setOption(T channel, Entry<? extends SocketOption<?>, ?> optEntry) throws
IOException {
- return setOptions(channel, ToolMapUtils.putAll(new
HashMap<SocketOption<?>, Object>(1), optEntry));
- }
-
- @SuppressWarnings({ "unchecked" })
- public static <T extends SelectableChannel & NetworkChannel> T
setOptions(T channel, Map<? extends SocketOption<?>, ?> optMap) throws
IOException {
- for (SocketOption<?> opt : optMap.keySet()) {
- if (ToolClassUtils.isAssignable(opt.getClass(),
ToolSocketOption.class)) {
- if (opt == ChannelSocketOptions.CHANNEL_BLOCKING) {
-
channel.configureBlocking(ChannelSocketOptions.CHANNEL_BLOCKING.type().cast(optMap.get(opt)));
- }
- } else {
- channel.setOption(((SocketOption<Object>) opt),
optMap.get(opt));
- }
- }
-
- return channel;
- }
-}
=======================================
--- /dcdt-core/src/main/resources/META-INF/spring/spring-core-nio.xml Sat
Feb 22 16:28:53 2014 UTC
+++ /dev/null
@@ -1,30 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<beans:beans
- xmlns:aop="http://www.springframework.org/schema/aop"
- xmlns:beans="http://www.springframework.org/schema/beans"
- xmlns:c="http://www.springframework.org/schema/c"
- xmlns:context="http://www.springframework.org/schema/context"
- xmlns:dcdt="http://direct-test.com/spring/schema/spring-dcdt"
- xmlns:p="http://www.springframework.org/schema/p"
- xmlns:task="http://www.springframework.org/schema/task"
- xmlns:util="http://www.springframework.org/schema/util"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="
- http://direct-test.com/spring/schema/spring-dcdt
http://direct-test.com/spring/schema/spring-dcdt.xsd
- http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop.xsd
- http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
- http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
- http://www.springframework.org/schema/task
http://www.springframework.org/schema/task/spring-task.xsd
- http://www.springframework.org/schema/util
http://www.springframework.org/schema/util/spring-util.xsd">
-
-
<!--====================================================================================================
- = CHANNELS
-
=====================================================================================================-->
- <beans:bean id="channel" class="java.nio.channels.SelectableChannel"
abstract="true"/>
-
- <beans:bean id="channelUdp" class="java.nio.channels.DatagramChannel"
parent="channel" factory-method="open" lazy-init="true" scope="prototype"/>
-
- <beans:bean id="channelTcp" class="java.nio.channels.SocketChannel"
parent="channel" factory-method="open" lazy-init="true" scope="prototype"/>
-
- <beans:bean id="channelTcpListen"
class="java.nio.channels.ServerSocketChannel" parent="channel"
factory-method="open" lazy-init="true" scope="prototype"/>
-</beans:beans>
=======================================
***Additional files exist in this changeset.***
Reply all
Reply to author
Forward
0 new messages