dorkbox/Network
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Code polish

Browse Source
This commit is contained in:
nathan 2018-01-11 23:44:17 +01:00
parent 71111c4e3c
commit ad7e0f44f9
5 changed files with 92 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/dorkbox/network/Client.java
View File

@ -65,7 +65,7 @@ class Client<C extends Connection> extends EndPointClient<C> implements Connecti
*/
public static
String getVersion() {
return "2.4";
return "2.5";
}
private final String localChannelName;

23
src/dorkbox/network/Server.java
View File

@ -18,8 +18,6 @@ package dorkbox.network;
import java.io.IOException;
import java.net.Socket;
import org.slf4j.Logger;
import dorkbox.network.connection.Connection;
import dorkbox.network.connection.EndPointServer;
import dorkbox.network.connection.registration.local.RegistrationLocalHandlerServer;
@ -66,7 +64,7 @@ class Server<C extends Connection> extends EndPointServer<C> {
*/
public static
String getVersion() {
return "2.4";
return "2.5";
}
/**
@ -305,22 +303,21 @@ class Server<C extends Connection> extends EndPointServer<C> {
ChannelFuture future;
// LOCAL
Logger logger2 = logger;
if (localBootstrap != null) {
try {
future = localBootstrap.bind();
future.await();
} catch (InterruptedException e) {
String errorMessage = stopWithErrorMessage(logger2, "Could not bind to LOCAL address on the server.", e);
String errorMessage = stopWithErrorMessage(logger, "Could not bind to LOCAL address on the server.", e);
throw new IllegalArgumentException(errorMessage);
}
if (!future.isSuccess()) {
String errorMessage = stopWithErrorMessage(logger2, "Could not bind to LOCAL address on the server.", future.cause());
String errorMessage = stopWithErrorMessage(logger, "Could not bind to LOCAL address on the server.", future.cause());
throw new IllegalArgumentException(errorMessage);
}
logger2.info("Listening on LOCAL address: '{}'", localChannelName);
logger.info("Listening on LOCAL address: '{}'", localChannelName);
manageForShutdown(future);
}
@ -332,7 +329,7 @@ class Server<C extends Connection> extends EndPointServer<C> {
future = tcpBootstrap.bind();
future.await();
} catch (Exception e) {
String errorMessage = stopWithErrorMessage(logger2,
String errorMessage = stopWithErrorMessage(logger,
"Could not bind to address " + hostName + " TCP port " + tcpPort +
" on the server.",
e);
@ -340,14 +337,14 @@ class Server<C extends Connection> extends EndPointServer<C> {
}
if (!future.isSuccess()) {
String errorMessage = stopWithErrorMessage(logger2,
String errorMessage = stopWithErrorMessage(logger,
"Could not bind to address " + hostName + " TCP port " + tcpPort +
" on the server.",
future.cause());
throw new IllegalArgumentException(errorMessage);
}
logger2.info("Listening on address {} at TCP port: {}", hostName, tcpPort);
logger.info("Listening on address {} at TCP port: {}", hostName, tcpPort);
manageForShutdown(future);
}
@ -358,7 +355,7 @@ class Server<C extends Connection> extends EndPointServer<C> {
future = udpBootstrap.bind();
future.await();
} catch (Exception e) {
String errorMessage = stopWithErrorMessage(logger2,
String errorMessage = stopWithErrorMessage(logger,
"Could not bind to address " + hostName + " UDP port " + udpPort +
" on the server.",
e);
@ -366,14 +363,14 @@ class Server<C extends Connection> extends EndPointServer<C> {
}
if (!future.isSuccess()) {
String errorMessage = stopWithErrorMessage(logger2,
String errorMessage = stopWithErrorMessage(logger,
"Could not bind to address " + hostName + " UDP port " + udpPort +
" on the server.",
future.cause());
throw new IllegalArgumentException(errorMessage);
}
logger2.info("Listening on address {} at UDP port: {}", hostName, udpPort);
logger.info("Listening on address {} at UDP port: {}", hostName, udpPort);
manageForShutdown(future);
}

48
src/dorkbox/network/connection/EndPoint.java
View File

@ -105,10 +105,10 @@ class EndPoint {
.getThreadGroup(), type.getSimpleName() + " " + THREADGROUP_NAME);
threadGroup.setDaemon(true);
this.logger = org.slf4j.LoggerFactory.getLogger(type.getSimpleName());
logger = org.slf4j.LoggerFactory.getLogger(type.getSimpleName());
this.shutdownHook = new Thread() {
shutdownHook = new Thread() {
@Override
public
void run() {
@ -117,10 +117,10 @@ class EndPoint {
}
}
};
this.shutdownHook.setName(shutdownHookName);
shutdownHook.setName(shutdownHookName);
try {
Runtime.getRuntime()
.addShutdownHook(this.shutdownHook);
.addShutdownHook(shutdownHook);
} catch (Throwable ignored) {
// if we are in the middle of shutdown, we cannot do this.
}
@ -131,8 +131,8 @@ class EndPoint {
*/
protected final
void manageForShutdown(ChannelFuture future) {
synchronized (this.shutdownChannelList) {
this.shutdownChannelList.add(future);
synchronized (shutdownChannelList) {
shutdownChannelList.add(future);
}
}
@ -141,8 +141,8 @@ class EndPoint {
*/
protected final
void manageForShutdown(EventLoopGroup loopGroup) {
synchronized (this.eventLoopGroups) {
this.eventLoopGroups.add(loopGroup);
synchronized (eventLoopGroups) {
eventLoopGroups.add(loopGroup);
}
}
@ -150,7 +150,7 @@ class EndPoint {
void shutdownChannels() {
synchronized (shutdownChannelList) {
// now we stop all of our channels
for (ChannelFuture f : this.shutdownChannelList) {
for (ChannelFuture f : shutdownChannelList) {
Channel channel = f.channel();
if (channel.isOpen()) {
channel.close()
@ -160,19 +160,19 @@ class EndPoint {
}
// we have to clear the shutdown list. (
this.shutdownChannelList.clear();
shutdownChannelList.clear();
}
}
protected final
String stopWithErrorMessage(Logger logger2, String errorMessage, Throwable throwable) {
if (logger2.isDebugEnabled() && throwable != null) {
String stopWithErrorMessage(Logger logger, String errorMessage, Throwable throwable) {
if (logger.isDebugEnabled() && throwable != null) {
// extra info if debug is enabled
logger2.error(errorMessage, throwable.getCause());
logger.error(errorMessage, throwable.getCause());
}
else {
logger2.error(errorMessage);
logger.error(errorMessage);
}
stop();
@ -199,7 +199,7 @@ class EndPoint {
public final
void stop() {
// only permit us to "stop" once!
if (!this.stopCalled.compareAndSet(false, true)) {
if (!stopCalled.compareAndSet(false, true)) {
return;
}
@ -220,7 +220,7 @@ class EndPoint {
// we have to make sure always run this from within it's OWN thread -- because if it's run from within
// a client/server thread executor, it will deadlock while waiting for the threadpool to terminate.
boolean isInEventLoop = false;
for (EventLoopGroup loopGroup : this.eventLoopGroups) {
for (EventLoopGroup loopGroup : eventLoopGroups) {
Iterator<EventExecutor> iterator = loopGroup.iterator();
while (iterator.hasNext()) {
EventExecutor next = iterator.next();
@ -276,11 +276,11 @@ class EndPoint {
void stopInThread() {
// make sure we are not trying to stop during a startup procedure.
// This will wait until we have finished starting up/shutting down.
synchronized (this.shutdownInProgress) {
synchronized (shutdownInProgress) {
// we want to WAIT until after the event executors have completed shutting down.
List<Future<?>> shutdownThreadList = new LinkedList<Future<?>>();
for (EventLoopGroup loopGroup : this.eventLoopGroups) {
for (EventLoopGroup loopGroup : eventLoopGroups) {
shutdownThreadList.add(loopGroup.shutdownGracefully(maxShutdownWaitTimeInMilliSeconds,
maxShutdownWaitTimeInMilliSeconds * 4,
TimeUnit.MILLISECONDS));
@ -298,7 +298,7 @@ class EndPoint {
shutdownChannels();
this.logger.info("Stopping endpoint.");
logger.info("Stopping endpoint.");
// there is no need to call "stop" again if we close the connection.
// however, if this is called WHILE from the shutdown hook, blammo! problems!
@ -310,7 +310,7 @@ class EndPoint {
.equals(shutdownHookName)) {
try {
Runtime.getRuntime()
.removeShutdownHook(this.shutdownHook);
.removeShutdownHook(shutdownHook);
} catch (Exception e) {
// ignore
}
@ -326,7 +326,7 @@ class EndPoint {
}
// tell the blocked "bind" method that it may continue (and exit)
this.blockUntilDone.countDown();
blockUntilDone.countDown();
}
/**
@ -336,9 +336,9 @@ class EndPoint {
void waitForShutdown() {
// we now BLOCK until the stop method is called.
try {
this.blockUntilDone.await();
blockUntilDone.await();
} catch (InterruptedException e) {
this.logger.error("Thread interrupted while waiting for stop!");
logger.error("Thread interrupted while waiting for stop!");
}
}
@ -350,6 +350,6 @@ class EndPoint {
public
String getName() {
return this.type.getSimpleName();
return type.getSimpleName();
}
}

104
src/dorkbox/network/connection/EndPointBase.java
View File

@ -133,9 +133,9 @@ class EndPointBase<C extends Connection> extends EndPoint {
// serialization stuff
if (config.serialization != null) {
this.serializationManager = config.serialization;
serializationManager = config.serialization;
} else {
this.serializationManager = CryptoSerializationManager.DEFAULT();
serializationManager = CryptoSerializationManager.DEFAULT();
}
// setup our RMI serialization managers. Can only be called once
@ -146,31 +146,31 @@ class EndPointBase<C extends Connection> extends EndPoint {
// The registration wrapper permits the registration process to access protected/package fields/methods, that we don't want
// to expose to external code. "this" escaping can be ignored, because it is benign.
//noinspection ThisEscapedInObjectConstruction
this.registrationWrapper = new RegistrationWrapper(this,
this.logger,
new KryoEncoder(this.serializationManager),
new KryoEncoderCrypto(this.serializationManager));
registrationWrapper = new RegistrationWrapper(this,
logger,
new KryoEncoder(serializationManager),
new KryoEncoderCrypto(serializationManager));
// we have to be able to specify WHAT property store we want to use, since it can change!
if (config.settingsStore == null) {
this.propertyStore = new PropertyStore();
propertyStore = new PropertyStore();
}
else {
this.propertyStore = config.settingsStore;
propertyStore = config.settingsStore;
}
this.propertyStore.init(this.serializationManager, null);
propertyStore.init(serializationManager, null);
// null it out, since it is sensitive!
config.settingsStore = null;
if (!(this.propertyStore instanceof NullSettingsStore)) {
if (!(propertyStore instanceof NullSettingsStore)) {
// initialize the private/public keys used for negotiating ECC handshakes
// these are ONLY used for IP connections. LOCAL connections do not need a handshake!
ECPrivateKeyParameters privateKey = this.propertyStore.getPrivateKey();
ECPublicKeyParameters publicKey = this.propertyStore.getPublicKey();
ECPrivateKeyParameters privateKey = propertyStore.getPrivateKey();
ECPublicKeyParameters publicKey = propertyStore.getPublicKey();
if (privateKey == null || publicKey == null) {
try {
@ -179,20 +179,20 @@ class EndPointBase<C extends Connection> extends EndPoint {
SecureRandom secureRandom = new SecureRandom(seedBytes);
secureRandom.nextBytes(seedBytes);
this.logger.debug("Now generating ECC (" + CryptoECC.curve25519 + ") keys. Please wait!");
logger.debug("Now generating ECC (" + CryptoECC.curve25519 + ") keys. Please wait!");
AsymmetricCipherKeyPair generateKeyPair = CryptoECC.generateKeyPair(CryptoECC.curve25519, secureRandom);
privateKey = (ECPrivateKeyParameters) generateKeyPair.getPrivate();
publicKey = (ECPublicKeyParameters) generateKeyPair.getPublic();
// save to properties file
this.propertyStore.savePrivateKey(privateKey);
this.propertyStore.savePublicKey(publicKey);
propertyStore.savePrivateKey(privateKey);
propertyStore.savePublicKey(publicKey);
this.logger.debug("Done with ECC keys!");
logger.debug("Done with ECC keys!");
} catch (Exception e) {
String message = "Unable to initialize/generate ECC keys. FORCED SHUTDOWN.";
this.logger.error(message);
logger.error(message);
throw new InitializationException(message);
}
}
@ -206,21 +206,21 @@ class EndPointBase<C extends Connection> extends EndPoint {
}
this.secureRandom = new SecureRandom(this.propertyStore.getSalt());
secureRandom = new SecureRandom(propertyStore.getSalt());
// we don't care about un-instantiated/constructed members, since the class type is the only interest.
this.connectionManager = new ConnectionManager<C>(type.getSimpleName(), connection0(null).getClass());
connectionManager = new ConnectionManager<C>(type.getSimpleName(), connection0(null).getClass());
// add the ping listener (internal use only!)
this.connectionManager.add(new PingSystemListener());
connectionManager.add(new PingSystemListener());
if (this.rmiEnabled) {
if (rmiEnabled) {
// these register the listener for registering a class implementation for RMI (internal use only)
this.connectionManager.add(new RegisterRmiSystemListener());
this.globalRmiBridge = new RmiBridge(logger, config.rmiExecutor, true);
connectionManager.add(new RegisterRmiSystemListener());
globalRmiBridge = new RmiBridge(logger, config.rmiExecutor, true);
}
else {
this.globalRmiBridge = null;
globalRmiBridge = null;
}
serializationManager.finishInit();
@ -231,14 +231,12 @@ class EndPointBase<C extends Connection> extends EndPoint {
*/
public
void disableRemoteKeyValidation() {
Logger logger2 = this.logger;
if (isConnected()) {
logger2.error("Cannot disable the remote key validation after this endpoint is connected!");
logger.error("Cannot disable the remote key validation after this endpoint is connected!");
}
else {
logger2.info("WARNING: Disabling remote key validation is a security risk!!");
this.disableRemoteKeyValidation = true;
logger.info("WARNING: Disabling remote key validation is a security risk!!");
disableRemoteKeyValidation = true;
}
}
@ -249,7 +247,7 @@ class EndPointBase<C extends Connection> extends EndPoint {
@SuppressWarnings("unchecked")
public
<S extends SettingsStore> S getPropertyStore() {
return (S) this.propertyStore;
return (S) propertyStore;
}
/**
@ -267,7 +265,7 @@ class EndPointBase<C extends Connection> extends EndPoint {
*/
public
int getIdleTimeout() {
return this.idleTimeoutMs;
return idleTimeoutMs;
}
/**
@ -278,7 +276,7 @@ class EndPointBase<C extends Connection> extends EndPoint {
*/
public
void setIdleTimeout(int idleTimeoutMs) {
this.idleTimeoutMs = idleTimeoutMs;
idleTimeoutMs = idleTimeoutMs;
}
/**
@ -288,7 +286,7 @@ class EndPointBase<C extends Connection> extends EndPoint {
*/
public final
boolean isConnected() {
return this.isConnected.get();
return isConnected.get();
}
/**
@ -296,7 +294,7 @@ class EndPointBase<C extends Connection> extends EndPoint {
*/
public
dorkbox.network.util.CryptoSerializationManager getSerialization() {
return this.serializationManager;
return serializationManager;
}
/**
@ -344,7 +342,7 @@ class EndPointBase<C extends Connection> extends EndPoint {
}
else {
if (this instanceof EndPointServer) {
wrapper = new ChannelNetworkWrapper(metaChannel, this.registrationWrapper);
wrapper = new ChannelNetworkWrapper(metaChannel, registrationWrapper);
}
else {
wrapper = new ChannelNetworkWrapper(metaChannel, null);
@ -352,7 +350,7 @@ class EndPointBase<C extends Connection> extends EndPoint {
}
// now initialize the connection channels with whatever extra info they might need.
connection.init(wrapper, (ConnectionManager<Connection>) this.connectionManager);
connection.init(wrapper, (ConnectionManager<Connection>) connectionManager);
if (rmiBridge != null) {
// notify our remote object space that it is able to receive method calls.
@ -378,12 +376,12 @@ class EndPointBase<C extends Connection> extends EndPoint {
*/
@SuppressWarnings("unchecked")
void connectionConnected0(ConnectionImpl connection) {
this.isConnected.set(true);
isConnected.set(true);
// prep the channel wrapper
connection.prep();
this.connectionManager.onConnected((C) connection);
connectionManager.onConnected((C) connection);
}
/**
@ -391,7 +389,7 @@ class EndPointBase<C extends Connection> extends EndPoint {
*/
public final
Listeners listeners() {
return this.connectionManager;
return connectionManager;
}
/**
@ -399,7 +397,7 @@ class EndPointBase<C extends Connection> extends EndPoint {
*/
public
List<C> getConnections() {
return this.connectionManager.getConnections();
return connectionManager.getConnections();
}
/**
@ -408,7 +406,7 @@ class EndPointBase<C extends Connection> extends EndPoint {
@SuppressWarnings("unchecked")
public
Collection<C> getConnectionsAs() {
return this.connectionManager.getConnections();
return connectionManager.getConnections();
}
/**
@ -430,12 +428,12 @@ class EndPointBase<C extends Connection> extends EndPoint {
Thread.yield();
// stop does the same as this + more
this.connectionManager.closeConnections();
connectionManager.closeConnections();
// Sometimes there might be "lingering" connections (ie, halfway though registration) that need to be closed.
this.registrationWrapper.closeChannels(maxShutdownWaitTimeInMilliSeconds);
registrationWrapper.closeChannels(maxShutdownWaitTimeInMilliSeconds);
this.isConnected.set(false);
isConnected.set(false);
}
/**
@ -447,7 +445,7 @@ class EndPointBase<C extends Connection> extends EndPoint {
protected
boolean shouldShutdownHookRun() {
// connectionManager.shutdown accurately reflects the state of the app. Safe to use here
return (this.connectionManager != null && !this.connectionManager.shutdown.get());
return (connectionManager != null && !connectionManager.shutdown.get());
}
@Override
@ -456,14 +454,14 @@ class EndPointBase<C extends Connection> extends EndPoint {
closeConnections();
// this does a closeConnections + clear_listeners
this.connectionManager.stop();
connectionManager.stop();
}
@Override
protected
void stopExtraActionsInternal() {
// shutdown the database store
this.propertyStore.close();
propertyStore.close();
}
@Override
@ -471,8 +469,8 @@ class EndPointBase<C extends Connection> extends EndPoint {
int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + (this.privateKey == null ? 0 : this.privateKey.hashCode());
result = prime * result + (this.publicKey == null ? 0 : this.publicKey.hashCode());
result = prime * result + (privateKey == null ? 0 : privateKey.hashCode());
result = prime * result + (publicKey == null ? 0 : publicKey.hashCode());
return result;
}
@ -491,20 +489,20 @@ class EndPointBase<C extends Connection> extends EndPoint {
}
EndPointBase other = (EndPointBase) obj;
if (this.privateKey == null) {
if (privateKey == null) {
if (other.privateKey != null) {
return false;
}
}
else if (!CryptoECC.compare(this.privateKey, other.privateKey)) {
else if (!CryptoECC.compare(privateKey, other.privateKey)) {
return false;
}
if (this.publicKey == null) {
if (publicKey == null) {
if (other.publicKey != null) {
return false;
}
}
else if (!CryptoECC.compare(this.publicKey, other.publicKey)) {
else if (!CryptoECC.compare(publicKey, other.publicKey)) {
return false;
}
return true;

16
src/dorkbox/network/connection/EndPointClient.java
View File

@ -20,8 +20,6 @@ import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import org.slf4j.Logger;
import dorkbox.network.Client;
import dorkbox.network.Configuration;
import dorkbox.network.connection.bridge.ConnectionBridge;
@ -95,7 +93,6 @@ class EndPointClient<C extends Connection> extends EndPointBase<C> implements Ru
bootstrapWrapper.bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.connectionTimeout);
}
Logger logger2 = this.logger;
try {
// UDP : When this is CONNECT, a udp socket will ONLY accept UDP traffic from the remote address (ip/port combo).
// If the reply isn't from the correct port, then the other end will receive a "Port Unreachable" exception.
@ -103,7 +100,7 @@ class EndPointClient<C extends Connection> extends EndPointBase<C> implements Ru
future = bootstrapWrapper.bootstrap.connect();
future.await();
} catch (Exception e) {
String errorMessage = stopWithErrorMessage(logger2,
String errorMessage = stopWithErrorMessage(this.logger,
"Could not connect to the " + bootstrapWrapper.type + " server at " +
bootstrapWrapper.address + " on port: " + bootstrapWrapper.port,
e);
@ -111,15 +108,15 @@ class EndPointClient<C extends Connection> extends EndPointBase<C> implements Ru
}
if (!future.isSuccess()) {
String errorMessage = stopWithErrorMessage(logger2,
String errorMessage = stopWithErrorMessage(this.logger,
"Could not connect to the " + bootstrapWrapper.type + " server at " +
bootstrapWrapper.address + " on port: " + bootstrapWrapper.port,
future.cause());
throw new IllegalArgumentException(errorMessage);
}
if (logger2.isTraceEnabled()) {
logger2.trace("Waiting for registration from server.");
if (this.logger.isTraceEnabled()) {
this.logger.trace("Waiting for registration from server.");
}
manageForShutdown(future);
}
@ -144,9 +141,8 @@ class EndPointClient<C extends Connection> extends EndPointBase<C> implements Ru
}
Logger logger2 = this.logger;
if (logger2.isTraceEnabled()) {
logger2.trace("Registered protocol from server.");
if (this.logger.isTraceEnabled()) {
this.logger.trace("Registered protocol from server.");
}
// only let us continue with connections (this starts up the client/server implementations) once ALL of the