[teiid-commits] teiid SVN: r2482 - in branches/7.1.x: client/src/main/java/org/teiid/client/security and 6 other directories.
teiid-commits at lists.jboss.org
teiid-commits at lists.jboss.org
Fri Aug 20 18:18:48 EDT 2010
Author: shawkins
Date: 2010-08-20 18:18:46 -0400 (Fri, 20 Aug 2010)
New Revision: 2482
Modified:
branches/7.1.x/client/src/main/java/org/teiid/adminapi/AdminFactory.java
branches/7.1.x/client/src/main/java/org/teiid/client/security/ILogon.java
branches/7.1.x/client/src/main/java/org/teiid/jdbc/ConnectionImpl.java
branches/7.1.x/client/src/main/java/org/teiid/jdbc/XAConnectionImpl.java
branches/7.1.x/client/src/main/java/org/teiid/net/HostInfo.java
branches/7.1.x/client/src/main/java/org/teiid/net/ServerConnection.java
branches/7.1.x/client/src/main/java/org/teiid/net/socket/AdminApiServerDiscovery.java
branches/7.1.x/client/src/main/java/org/teiid/net/socket/Handshake.java
branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerConnection.java
branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerConnectionFactory.java
branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstance.java
branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstanceFactory.java
branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstanceImpl.java
branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestAdminApiServerDiscovery.java
branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestSocketServerConnection.java
branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestSocketServerInstanceImpl.java
branches/7.1.x/runtime/src/main/java/org/teiid/transport/LocalServerConnection.java
branches/7.1.x/runtime/src/main/java/org/teiid/transport/LogonImpl.java
branches/7.1.x/runtime/src/main/java/org/teiid/transport/PgBackendProtocol.java
branches/7.1.x/runtime/src/main/java/org/teiid/transport/PgFrontendProtocol.java
branches/7.1.x/runtime/src/main/java/org/teiid/transport/SSLAwareChannelHandler.java
branches/7.1.x/runtime/src/main/java/org/teiid/transport/SocketClientInstance.java
branches/7.1.x/runtime/src/main/java/org/teiid/transport/SocketListener.java
branches/7.1.x/runtime/src/test/java/org/teiid/transport/TestCommSockets.java
branches/7.1.x/runtime/src/test/java/org/teiid/transport/TestSocketRemoting.java
Log:
TEIID-1211 TEIID-1204 reimplementing loadbalancing and failover logic. this initial checkin mainly addresses failover
Modified: branches/7.1.x/client/src/main/java/org/teiid/adminapi/AdminFactory.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/adminapi/AdminFactory.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/adminapi/AdminFactory.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -27,11 +27,9 @@
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Properties;
-import java.util.concurrent.ExecutionException;
import org.teiid.client.security.LogonException;
import org.teiid.client.util.ExceptionUtil;
-import org.teiid.client.util.ResultsFuture;
import org.teiid.core.TeiidRuntimeException;
import org.teiid.core.util.PropertiesUtils;
import org.teiid.net.CommunicationException;
@@ -51,42 +49,23 @@
private static final int DEFAULT_BOUNCE_WAIT = 2000;
- private final class ReconnectingProxy implements InvocationHandler {
+ private final class AdminProxy implements InvocationHandler {
private Admin target;
private ServerConnection registry;
private Properties p;
private boolean closed;
- public ReconnectingProxy(Properties p) throws ConnectionException, CommunicationException {
+ public AdminProxy(Properties p) throws ConnectionException, CommunicationException {
this.p = p;
this.registry = serverConnectionFactory.getConnection(p);
this.target = registry.getService(Admin.class);
}
- private synchronized Admin getTarget() throws AdminComponentException, CommunicationException {
+ private synchronized Admin getTarget() throws AdminComponentException {
if (closed) {
throw new AdminComponentException(NetPlugin.Util.getString("ERR.014.001.0001")); //$NON-NLS-1$
}
- if (target != null) {
- ResultsFuture<?> ping = registry.isOpen();
- if (ping != null) {
- try {
- ping.get();
- return target;
- } catch (InterruptedException e) {
- throw new CommunicationException(e);
- } catch (ExecutionException e) {
- //assume recoverable
- }
- }
- }
- try {
- registry = serverConnectionFactory.getConnection(p);
- } catch (ConnectionException e) {
- throw new AdminComponentException(e);
- }
- target = registry.getService(Admin.class);
return target;
}
@@ -97,7 +76,6 @@
close();
return null;
}
- Throwable t = null;
try {
return method.invoke(getTarget(), args);
} catch (InvocationTargetException e) {
@@ -109,10 +87,7 @@
}
}
throw e.getTargetException();
- } catch (CommunicationException e) {
- t = e;
}
- throw t;
}
public synchronized void close() {
@@ -231,7 +206,7 @@
p.setProperty(TeiidURL.CONNECTION.ADMIN, Boolean.TRUE.toString());
try {
- Admin serverAdmin = (Admin)Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] { Admin.class }, new ReconnectingProxy(p));
+ Admin serverAdmin = (Admin)Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] { Admin.class }, new AdminProxy(p));
return serverAdmin;
} catch (ConnectionException e) {
throw new AdminComponentException(e);
Modified: branches/7.1.x/client/src/main/java/org/teiid/client/security/ILogon.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/client/security/ILogon.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/client/security/ILogon.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -21,11 +21,13 @@
*/
package org.teiid.client.security;
+import java.util.Collection;
import java.util.Properties;
import org.teiid.client.util.ResultsFuture;
import org.teiid.core.ComponentNotFoundException;
import org.teiid.core.TeiidComponentException;
+import org.teiid.net.CommunicationException;
/**
@@ -33,7 +35,7 @@
*/
public interface ILogon {
LogonResult logon(Properties connectionProperties)
- throws LogonException, TeiidComponentException;
+ throws LogonException, TeiidComponentException, CommunicationException;
/**
* Ping the server to see if the client-server connection is alive.
@@ -41,8 +43,10 @@
* @throws ComponentNotFoundException if can't find the Session service.
*/
ResultsFuture<?> ping()
- throws InvalidSessionException, TeiidComponentException;
+ throws InvalidSessionException, TeiidComponentException, CommunicationException;
+ ResultsFuture<?> ping(Collection<String> sessions)
+ throws TeiidComponentException, CommunicationException;
/**
* Log off the specified session.
@@ -51,5 +55,5 @@
*/
ResultsFuture<?> logoff() throws InvalidSessionException, TeiidComponentException;
- void assertIdentity(SessionToken sessionId) throws InvalidSessionException, TeiidComponentException;
+ void assertIdentity(SessionToken sessionId) throws InvalidSessionException, TeiidComponentException, CommunicationException;
}
Modified: branches/7.1.x/client/src/main/java/org/teiid/jdbc/ConnectionImpl.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/jdbc/ConnectionImpl.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/jdbc/ConnectionImpl.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -45,9 +45,6 @@
import java.util.List;
import java.util.Map;
import java.util.Properties;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -878,20 +875,7 @@
}
public boolean isValid(int timeout) throws SQLException {
- ResultsFuture<?> future = this.getServerConnection().isOpen();
- if (future == null) {
- return false;
- }
- try {
- future.get(timeout, TimeUnit.SECONDS);
- return true;
- } catch (InterruptedException e) {
- return false;
- } catch (ExecutionException e) {
- return false;
- } catch (TimeoutException e) {
- return false;
- }
+ return this.getServerConnection().isOpen(timeout * 1000);
}
public void recycleConnection() {
@@ -918,7 +902,7 @@
//perform load balancing
if (this.serverConn instanceof SocketServerConnection) {
- ((SocketServerConnection)this.serverConn).selectNewServerInstance(this.getDQP());
+ ((SocketServerConnection)this.serverConn).selectNewServerInstance();
}
}
Modified: branches/7.1.x/client/src/main/java/org/teiid/jdbc/XAConnectionImpl.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/jdbc/XAConnectionImpl.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/jdbc/XAConnectionImpl.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -80,7 +80,7 @@
ex = ExceptionUtil.getExceptionOfType(e, CommunicationException.class);
if (ex instanceof SingleInstanceCommunicationException) {
ServerConnection sc = proxiedConnection.getServerConnection();
- if (sc.isOpen() != null) {
+ if (!sc.isOpen(ServerConnection.PING_INTERVAL)) {
ex = null;
}
}
Modified: branches/7.1.x/client/src/main/java/org/teiid/net/HostInfo.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/net/HostInfo.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/net/HostInfo.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -23,6 +23,7 @@
package org.teiid.net;
import java.net.InetAddress;
+import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import org.teiid.core.util.ArgCheck;
@@ -38,12 +39,12 @@
private String hostName;
private int portNumber = 0;
private InetAddress inetAddress;
-
- public InetAddress getInetAddress() throws UnknownHostException {
- if (inetAddress != null) {
- return inetAddress;
- }
- return InetAddress.getByName(this.hostName);
+ private boolean ssl;
+
+ public HostInfo(String hostName, InetSocketAddress addr) {
+ this.hostName = hostName;
+ this.portNumber = addr.getPort();
+ this.inetAddress = addr.getAddress();
}
public HostInfo (String host, int port) {
@@ -61,6 +62,13 @@
}
}
+ public InetAddress getInetAddress() throws UnknownHostException {
+ if (inetAddress != null) {
+ return inetAddress;
+ }
+ return InetAddress.getByName(this.hostName);
+ }
+
public String getHostName() {
return hostName;
}
@@ -87,7 +95,13 @@
return false;
}
HostInfo hostInfo = (HostInfo) obj;
- return hostName.equals(hostInfo.getHostName()) && portNumber == hostInfo.getPortNumber();
+ if (portNumber != hostInfo.getPortNumber()) {
+ return false;
+ }
+ if (inetAddress != null && hostInfo.inetAddress != null) {
+ return inetAddress.equals(hostInfo.inetAddress);
+ }
+ return hostName.equals(hostInfo.getHostName());
}
/**
@@ -98,5 +112,17 @@
int hc = HashCodeUtil.hashCode(0, hostName);
return HashCodeUtil.hashCode(hc, portNumber);
}
+
+ public boolean isResolved() {
+ return this.inetAddress != null;
+ }
+
+ public boolean isSsl() {
+ return ssl;
+ }
+
+ public void setSsl(boolean ssl) {
+ this.ssl = ssl;
+ }
}
Modified: branches/7.1.x/client/src/main/java/org/teiid/net/ServerConnection.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/net/ServerConnection.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/net/ServerConnection.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -23,7 +23,6 @@
package org.teiid.net;
import org.teiid.client.security.LogonResult;
-import org.teiid.client.util.ResultsFuture;
public interface ServerConnection {
@@ -33,7 +32,7 @@
void close();
- ResultsFuture<?> isOpen();
+ boolean isOpen(long msToTest);
LogonResult getLogonResult();
Modified: branches/7.1.x/client/src/main/java/org/teiid/net/socket/AdminApiServerDiscovery.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/net/socket/AdminApiServerDiscovery.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/net/socket/AdminApiServerDiscovery.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -82,7 +82,7 @@
&& (info.lastDiscoveryTime < System.currentTimeMillis() - DISCOVERY_TIMEOUT || info.knownHosts.isEmpty())) {
Admin serverAdmin = instance.getService(Admin.class);
try {
- Collection<ProcessObject> processes = serverAdmin.getProcesses("*");
+ Collection<ProcessObject> processes = serverAdmin.getProcesses("*"); //$NON-NLS-1$
info.knownHosts.clear();
for (ProcessObject processObject : processes) {
if (!processObject.isEnabled() || !processObject.isRunning()) {
Modified: branches/7.1.x/client/src/main/java/org/teiid/net/socket/Handshake.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/net/socket/Handshake.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/net/socket/Handshake.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -50,8 +50,8 @@
/**
* @param version The version to set.
*/
- public void setVersion(String version) {
- this.version = version;
+ public void setVersion() {
+ this.version = ApplicationInfo.getInstance().getMajorReleaseNumber();
}
/**
Modified: branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerConnection.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerConnection.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerConnection.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -26,22 +26,18 @@
package org.teiid.net.socket;
import java.io.IOException;
-import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
-import java.net.SocketAddress;
+import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -52,6 +48,7 @@
import org.teiid.client.util.ExceptionUtil;
import org.teiid.client.util.ResultsFuture;
import org.teiid.core.TeiidComponentException;
+import org.teiid.core.TeiidException;
import org.teiid.net.CommunicationException;
import org.teiid.net.ConnectionException;
import org.teiid.net.HostInfo;
@@ -66,8 +63,7 @@
*/
public class SocketServerConnection implements ServerConnection {
- private static final int RETRY_COUNT = 3;
-
+ private static final int FAILOVER_PING_INTERVAL = 1000;
private SocketServerInstanceFactory connectionFactory;
private ServerDiscovery serverDiscovery;
private static Logger log = Logger.getLogger("org.teiid.client.sockets"); //$NON-NLS-1$
@@ -76,72 +72,44 @@
private Properties connProps;
private SocketServerInstance serverInstance;
- private volatile LogonResult logonResult;
+ private LogonResult logonResult;
+ private Map<HostInfo, LogonResult> logonResults = new ConcurrentHashMap<HostInfo, LogonResult>();
private ILogon logon;
- private Timer pingTimer;
private boolean closed;
private boolean failOver;
+ private long lastPing = System.currentTimeMillis();
+ private int pingFailOverInterval = FAILOVER_PING_INTERVAL;
public SocketServerConnection(
SocketServerInstanceFactory connectionFactory, boolean secure,
- ServerDiscovery serverDiscovery, Properties connProps,
- Timer pingTimer) throws CommunicationException, ConnectionException {
+ ServerDiscovery serverDiscovery, Properties connProps) throws CommunicationException, ConnectionException {
this.connectionFactory = connectionFactory;
this.serverDiscovery = serverDiscovery;
this.connProps = connProps;
this.secure = secure;
+ //ILogon that is allowed to failover
this.logon = this.getService(ILogon.class);
this.failOver = Boolean.valueOf(connProps.getProperty(TeiidURL.CONNECTION.AUTO_FAILOVER)).booleanValue();
-
- authenticate();
-
- this.pingTimer = pingTimer;
- schedulePing();
+ this.failOver |= Boolean.valueOf(connProps.getProperty(TeiidURL.CONNECTION.ADMIN)).booleanValue();
+ selectServerInstance();
}
-
- private void schedulePing() {
- if (this.pingTimer != null) {
- this.pingTimer.schedule(new TimerTask() {
-
- private ResultsFuture<?> ping;
-
- @Override
- public void run() {
- if (ping == null) {
- ping = isOpen();
- }
- if (ping != null) {
- try {
- ping.get(1, TimeUnit.SECONDS);
- ping = null;
- return;
- } catch (TimeoutException e) {
- return;
- } catch (Throwable e) {
- handlePingError(e);
- }
- }
- this.cancel();
- }
-
- }, PING_INTERVAL, PING_INTERVAL);
- }
- }
/**
* Implements a sticky random selection policy
* TODO: make this customizable
* TODO: put more information on hostinfo as to process response time, last successful connect, etc.
+ * @throws ConnectionException
*/
public synchronized SocketServerInstance selectServerInstance()
- throws CommunicationException {
+ throws CommunicationException, ConnectionException {
if (closed) {
throw new CommunicationException(NetPlugin.Util.getString("SocketServerConnection.closed")); //$NON-NLS-1$
}
if (this.serverInstance != null && (!failOver || this.serverInstance.isOpen())) {
return this.serverInstance;
}
- List<HostInfo> hostKeys = new ArrayList<HostInfo>(this.serverDiscovery.getKnownHosts(logonResult, this.serverInstance));
+ List<HostInfo> hostKeys = new ArrayList<HostInfo>(this.serverDiscovery.getKnownHosts(logonResult, null));
+ boolean discoverHosts = true;
closeServerInstance();
List<HostInfo> hostCopy = new ArrayList<HostInfo>(hostKeys);
int knownHosts = hostKeys.size();
@@ -150,23 +118,38 @@
Exception ex = null;
try {
- SocketServerInstance instance = connectionFactory.getServerInstance(hostInfo, secure);
- this.serverInstance = instance;
- if (this.logonResult != null) {
- ILogon newLogon = instance.getService(ILogon.class);
- newLogon.assertIdentity(logonResult.getSessionToken());
+ ILogon newLogon = connect(hostInfo);
+ if (this.logonResult == null) {
+ try {
+ this.logonResult = newLogon.logon(connProps);
+ this.logonResults.put(this.serverInstance.getHostInfo(), this.logonResult);
+ this.connectionFactory.connected(this.serverInstance, this.logonResult.getSessionToken());
+ this.serverDiscovery.connectionSuccessful(hostInfo);
+ if (discoverHosts) {
+ List<HostInfo> updatedHosts = this.serverDiscovery.getKnownHosts(logonResult, this.serverInstance);
+ if (updatedHosts.size() > 1 && new HashSet<HostInfo>(updatedHosts).size() > new HashSet<HostInfo>(hostCopy).size()) {
+ hostKeys = updatedHosts;
+ closeServerInstance();
+ discoverHosts = false;
+ continue;
+ }
+ }
+ } catch (LogonException e) {
+ // Propagate the original message as it contains the message we want
+ // to give to the user
+ throw new ConnectionException(e, e.getMessage());
+ } catch (TeiidComponentException e) {
+ if (e.getCause() instanceof CommunicationException) {
+ throw (CommunicationException)e.getCause();
+ }
+ throw new CommunicationException(e, NetPlugin.Util.getString("PlatformServerConnectionFactory.Unable_to_find_a_component_used_in_logging_on_to")); //$NON-NLS-1$
+ }
}
- this.serverDiscovery.connectionSuccessful(hostInfo);
return this.serverInstance;
} catch (IOException e) {
ex = e;
- } catch (InvalidSessionException e) {
- shutdown(false);
- throw new CommunicationException(e,NetPlugin.Util.getString("SocketServerInstance.Connection_Error.Connect_Failed", hostInfo.getHostName(), String.valueOf(hostInfo.getPortNumber()), e.getMessage())); //$NON-NLS-1$
} catch (SingleInstanceCommunicationException e) {
ex = e;
- } catch (TeiidComponentException e) {
- ex = e;
}
this.serverDiscovery.markInstanceAsBad(hostInfo);
if (knownHosts == 1) { //just a single host, use the exception
@@ -179,133 +162,127 @@
}
throw new CommunicationException(NetPlugin.Util.getString("SocketServerInstancePool.No_valid_host_available", hostCopy.toString())); //$NON-NLS-1$
}
-
- public synchronized void authenticate() throws ConnectionException, CommunicationException {
- this.logonResult = null;
- // Log on to server
- try {
- this.logonResult = logon.logon(connProps);
- List<HostInfo> knownHosts = this.serverDiscovery.getKnownHosts(logonResult, this.serverInstance);
- if (knownHosts.size() > 1 && !new HashSet<HostInfo>(knownHosts).equals(new HashSet<HostInfo>(this.serverDiscovery.getKnownHosts(logonResult, null)))) {
- //if there are multiple instances, allow for load-balancing
- closeServerInstance();
- }
- return;
- } catch (LogonException e) {
- // Propagate the original message as it contains the message we want
- // to give to the user
- throw new ConnectionException(e, e.getMessage());
- } catch (TeiidComponentException e) {
- if (e.getCause() instanceof CommunicationException) {
- throw (CommunicationException)e.getCause();
- }
- throw new CommunicationException(e, NetPlugin.Util.getString("PlatformServerConnectionFactory.Unable_to_find_a_component_used_in_logging_on_to")); //$NON-NLS-1$
- }
- }
-
- class ServerConnectionInvocationHandler implements InvocationHandler {
-
- private Class<?> targetClass;
- private Object target;
-
- public ServerConnectionInvocationHandler(Class<?> targetClass) {
- this.targetClass = targetClass;
+
+ private ILogon connect(HostInfo hostInfo) throws CommunicationException,
+ IOException {
+ if (!hostInfo.isResolved()) {
+ hostInfo = new HostInfo(hostInfo.getHostName(), new InetSocketAddress(hostInfo.getInetAddress(), hostInfo.getPortNumber()));
}
-
- private synchronized Object getTarget() throws CommunicationException {
- if (this.target == null) {
- this.target = selectServerInstance().getService(targetClass);
+ hostInfo.setSsl(secure);
+ this.serverInstance = connectionFactory.getServerInstance(hostInfo);
+ this.logonResult = logonResults.get(hostInfo);
+ ILogon newLogon = this.serverInstance.getService(ILogon.class);
+ if (this.logonResult != null) {
+ try {
+ newLogon.assertIdentity(logonResult.getSessionToken());
+ } catch (TeiidException e) {
+ // session is no longer valid
+ disconnect();
}
- return this.target;
}
-
- public Object invoke(Object proxy, Method method, Object[] args)
- throws Throwable {
- Throwable exception = null;
- for (int i = 0; i < RETRY_COUNT; i++) {
- try {
- return method.invoke(getTarget(), args);
- } catch (InvocationTargetException t) {
- exception = t.getTargetException();
- } catch (Throwable t) {
- exception = t;
- }
- if (!failOver || ExceptionUtil.getExceptionOfType(exception, SingleInstanceCommunicationException.class) == null) {
- break;
- }
- invalidateTarget();
- //TODO: look for invalid session exception
- }
- throw ExceptionUtil.convertException(method, exception);
- }
-
- private synchronized void invalidateTarget() {
- this.target = null;
- }
-
+ return newLogon;
}
-
+
public <T> T getService(Class<T> iface) {
- return iface.cast(Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {iface}, new ServerConnectionInvocationHandler(iface)));
+ return iface.cast(Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {iface}, new SocketServerInstanceImpl.RemoteInvocationHandler(iface) {
+ @Override
+ protected SocketServerInstance getInstance() throws CommunicationException {
+ if (failOver && System.currentTimeMillis() - lastPing > pingFailOverInterval) {
+ try {
+ ResultsFuture<?> future = selectServerInstance().getService(ILogon.class).ping();
+ future.get();
+ } catch (SingleInstanceCommunicationException e) {
+ closeServerInstance();
+ } catch (CommunicationException e) {
+ throw e;
+ } catch (InvalidSessionException e) {
+ disconnect();
+ closeServerInstance();
+ } catch (Exception e) {
+ closeServerInstance();
+ }
+ }
+ lastPing = System.currentTimeMillis();
+ try {
+ return selectServerInstance();
+ } catch (ConnectionException e) {
+ throw new CommunicationException(e);
+ }
+ }
+
+ public Object invoke(Object proxy, Method method, Object[] args)
+ throws Throwable {
+ try {
+ return super.invoke(proxy, method, args);
+ } catch (Exception e) {
+ if (ExceptionUtil.getExceptionOfType(e, InvalidSessionException.class) != null) {
+ disconnect();
+ }
+ throw e;
+ }
+ }
+
+ }));
}
+
public synchronized void close() {
- shutdown(true);
- }
- private synchronized void shutdown(boolean logoff) {
if (this.closed) {
return;
}
- if (logoff) {
+ if (this.serverInstance != null) {
+ logoff();
+ }
+
+ for (Map.Entry<HostInfo, LogonResult> logonEntry : logonResults.entrySet()) {
try {
- //make a best effort to send the logoff
- Future<?> writeFuture = this.logon.logoff();
- writeFuture.get(5000, TimeUnit.MILLISECONDS);
- } catch (InvalidSessionException e) {
- //ignore
- } catch (InterruptedException e) {
- //ignore
- } catch (ExecutionException e) {
- //ignore
- } catch (TimeoutException e) {
- //ignore
- } catch (TeiidComponentException e) {
- //ignore
+ connect(logonEntry.getKey());
+ logoff();
+ } catch (Exception e) {
+
}
}
- closeServerInstance();
-
this.closed = true;
this.serverDiscovery.shutdown();
}
+
+ private void logoff() {
+ disconnect();
+ try {
+ //make a best effort to send the logoff
+ Future<?> writeFuture = this.serverInstance.getService(ILogon.class).logoff();
+ writeFuture.get(5000, TimeUnit.MILLISECONDS);
+ } catch (Exception e) {
+ //ignore
+ }
+ closeServerInstance();
+ }
+
+ private void disconnect() {
+ this.logonResults.remove(this.serverInstance.getHostInfo());
+ if (this.logonResult != null) {
+ this.connectionFactory.disconnected(this.serverInstance, this.logonResult.getSessionToken());
+ }
+ }
- public synchronized ResultsFuture<?> isOpen() {
+ private synchronized ResultsFuture<?> isOpen() throws CommunicationException, InvalidSessionException, TeiidComponentException {
if (this.closed) {
- return null;
+ throw new CommunicationException();
}
+ return logon.ping();
+ }
+
+ public boolean isOpen(long msToTest) {
try {
- if (!selectServerInstance().isOpen()) {
- return null;
- }
- } catch (CommunicationException e) {
- return null;
- }
- try {
- return logon.ping();
+ ResultsFuture<?> future = isOpen();
+ future.get(msToTest, TimeUnit.MILLISECONDS);
+ return true;
} catch (Throwable th) {
- return null;
+ return false;
}
}
- private void handlePingError(Throwable th) {
- if (ExceptionUtil.getExceptionOfType(th, InvalidSessionException.class) != null) {
- shutdown(false);
- } else {
- close();
- }
- }
-
public LogonResult getLogonResult() {
return logonResult;
}
@@ -321,20 +298,22 @@
if (!(otherService instanceof SocketServerConnection)) {
return false;
}
- SocketAddress address = selectServerInstance().getRemoteAddress();
- if (address == null) {
- return false;
+ try {
+ return selectServerInstance().getHostInfo().equals(((SocketServerConnection)otherService).selectServerInstance().getHostInfo());
+ } catch (ConnectionException e) {
+ throw new CommunicationException(e);
}
- return address.equals(((SocketServerConnection)otherService).selectServerInstance().getRemoteAddress());
}
- public void selectNewServerInstance(Object service) {
+ public void selectNewServerInstance() {
closeServerInstance();
- ServerConnectionInvocationHandler handler = (ServerConnectionInvocationHandler)Proxy.getInvocationHandler(service);
- handler.invalidateTarget();
}
public void setFailOver(boolean failOver) {
this.failOver = failOver;
}
+
+ public void setFailOverPingInterval(int pingFailOverInterval) {
+ this.pingFailOverInterval = pingFailOverInterval;
+ }
}
\ No newline at end of file
Modified: branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerConnectionFactory.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerConnectionFactory.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerConnectionFactory.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -30,11 +30,16 @@
import java.lang.reflect.Proxy;
import java.net.InetAddress;
import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Properties;
+import java.util.Set;
import java.util.Timer;
+import java.util.TimerTask;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -42,12 +47,15 @@
import java.util.logging.Logger;
import org.teiid.client.security.ILogon;
+import org.teiid.client.security.InvalidSessionException;
+import org.teiid.client.security.SessionToken;
import org.teiid.core.TeiidException;
import org.teiid.core.util.PropertiesUtils;
import org.teiid.core.util.ReflectionHelper;
import org.teiid.net.CommunicationException;
import org.teiid.net.ConnectionException;
import org.teiid.net.HostInfo;
+import org.teiid.net.ServerConnection;
import org.teiid.net.ServerConnectionFactory;
import org.teiid.net.TeiidURL;
@@ -106,13 +114,11 @@
private static class CachedInstance {
HostInfo info;
Integer instance;
- boolean ssl;
SocketServerInstance actual;
SocketServerInstance proxy;
- public CachedInstance(HostInfo info, boolean ssl) {
+ public CachedInstance(HostInfo info) {
this.info = info;
- this.ssl = ssl;
}
@Override
@@ -129,7 +135,7 @@
return false;
}
CachedInstance other = (CachedInstance) obj;
- if (!info.equals(other.info) || ssl != other.ssl) {
+ if (!info.equals(other.info)) {
return false;
}
if (instance == null || other.instance == null) {
@@ -142,6 +148,8 @@
private ObjectChannelFactory channelFactory;
private Timer pingTimer;
+ private HashMap<HostInfo, Set<SessionToken>> sessions = new HashMap<HostInfo, Set<SessionToken>>();
+
//instance pooling
private AtomicInteger instanceCount = new AtomicInteger();
private Map<CachedInstance, CachedInstance> instancePool = new LinkedHashMap<CachedInstance, CachedInstance>();
@@ -180,15 +188,58 @@
public void initialize(Properties info) {
PropertiesUtils.setBeanProperties(this, info, "org.teiid.sockets"); //$NON-NLS-1$
this.pingTimer = new Timer("SocketPing", true); //$NON-NLS-1$
+ this.pingTimer.schedule(new TimerTask() {
+
+ @Override
+ public void run() {
+ Set<Map.Entry<HostInfo, Set<SessionToken>>> sessionEntries = null;
+ synchronized (sessions) {
+ sessionEntries = new HashSet<Map.Entry<HostInfo, Set<SessionToken>>>(sessions.entrySet());
+ }
+ for (Map.Entry<HostInfo, Set<SessionToken>> entry : sessionEntries) {
+ SocketServerInstance instance = null;
+ HashSet<SessionToken> entries = null;
+ synchronized (sessions) {
+ entries = new HashSet<SessionToken>(entry.getValue());
+ }
+ try {
+ instance = getServerInstance(entry.getKey());
+ ILogon logon = instance.getService(ILogon.class);
+ if ("7.1.1".compareTo(instance.getServerVersion()) > 0) { //$NON-NLS-1$
+ for (SessionToken session : entries) {
+ try {
+ logon.assertIdentity(session);
+ logon.ping();
+ } catch (InvalidSessionException e) {
+ }
+ }
+ } else {
+ ArrayList<String> sessionStrings = new ArrayList<String>(entry.getValue().size());
+ for (SessionToken session : entries) {
+ sessionStrings.add(session.getSessionID());
+ }
+ logon.ping(sessionStrings);
+ }
+ } catch (Exception e) {
+ log.log(Level.WARNING, "Error performing keep-alive ping", e); //$NON-NLS-1$
+ } finally {
+ if (instance != null) {
+ instance.shutdown();
+ }
+ }
+ }
+ }
+ }, ServerConnection.PING_INTERVAL, ServerConnection.PING_INTERVAL);
this.channelFactory = new OioOjbectChannelFactory(info);
}
-
- public SocketServerInstance getServerInstance(HostInfo info, boolean ssl) throws CommunicationException, IOException {
+
+ @Override
+ public SocketServerInstance getServerInstance(HostInfo info) throws CommunicationException, IOException {
CachedInstance key = null;
- CachedInstance instance = null;
boolean useCache = this.maxCachedInstances > 0;
if (useCache) {
- key = new CachedInstance(info, ssl);
+ CachedInstance instance = null;
+ key = new CachedInstance(info);
synchronized (instancePool) {
instance = instancePool.remove(key);
}
@@ -197,7 +248,7 @@
boolean valid = false;
try {
Future<?> success = logon.ping();
- success.get(this.channelFactory.getSoTimeout(), TimeUnit.MICROSECONDS);
+ success.get(this.channelFactory.getSoTimeout(), TimeUnit.MILLISECONDS);
valid = true;
} catch (Exception e) {
log.log(Level.FINE, "Error performing ping, will select another instance", e); //$NON-NLS-1$
@@ -219,7 +270,7 @@
}
}
}
- SocketServerInstanceImpl ssii = new SocketServerInstanceImpl(info, ssl, getSynchronousTtl());
+ SocketServerInstanceImpl ssii = new SocketServerInstanceImpl(info, getSynchronousTtl());
ssii.connect(this.channelFactory);
if (useCache) {
key.actual = ssii;
@@ -256,7 +307,7 @@
discovery.init(url, connectionProperties);
- return new SocketServerConnection(this, url.isUsingSSL(), discovery, connectionProperties, pingTimer);
+ return new SocketServerConnection(this, url.isUsingSSL(), discovery, connectionProperties);
}
static void updateConnectionProperties(Properties connectionProperties) {
@@ -284,5 +335,30 @@
public void setMaxCachedInstances(int maxCachedInstances) {
this.maxCachedInstances = maxCachedInstances;
}
+
+ @Override
+ public void connected(SocketServerInstance instance, SessionToken session) {
+ synchronized (sessions) {
+ Set<SessionToken> instanceSessions = sessions.get(instance.getHostInfo());
+ if (instanceSessions == null) {
+ instanceSessions = new HashSet<SessionToken>();
+ sessions.put(instance.getHostInfo(), instanceSessions);
+ }
+ instanceSessions.add(session);
+ }
+ }
+
+ @Override
+ public void disconnected(SocketServerInstance instance, SessionToken session) {
+ synchronized (sessions) {
+ Set<SessionToken> instanceSessions = sessions.get(instance.getHostInfo());
+ if (instanceSessions != null) {
+ instanceSessions.remove(session);
+ if (instanceSessions.isEmpty()) {
+ sessions.remove(instance.getHostInfo());
+ }
+ }
+ }
+ }
}
Modified: branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstance.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstance.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstance.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -22,9 +22,14 @@
package org.teiid.net.socket;
-import java.net.SocketAddress;
+import java.io.Serializable;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.teiid.client.util.ResultsFuture;
+import org.teiid.client.util.ResultsReceiver;
import org.teiid.core.crypto.Cryptor;
+import org.teiid.net.CommunicationException;
import org.teiid.net.HostInfo;
@@ -33,8 +38,6 @@
<T> T getService(Class<T> iface);
void shutdown();
-
- SocketAddress getRemoteAddress();
HostInfo getHostInfo();
@@ -42,4 +45,12 @@
Cryptor getCryptor();
+ long getSynchTimeout();
+
+ void send(Message message, ResultsReceiver<Object> receiver, Serializable key) throws CommunicationException, InterruptedException;
+
+ void read(long timeout, TimeUnit unit, ResultsFuture<?> resultsFuture) throws TimeoutException, InterruptedException;
+
+ String getServerVersion();
+
}
\ No newline at end of file
Modified: branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstanceFactory.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstanceFactory.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstanceFactory.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -24,12 +24,17 @@
import java.io.IOException;
+import org.teiid.client.security.SessionToken;
import org.teiid.net.CommunicationException;
import org.teiid.net.HostInfo;
public interface SocketServerInstanceFactory {
- SocketServerInstance getServerInstance(HostInfo info, boolean ssl) throws CommunicationException, IOException;
+ SocketServerInstance getServerInstance(HostInfo info) throws CommunicationException, IOException;
+
+ void connected(SocketServerInstance instance, SessionToken session);
+
+ void disconnected(SocketServerInstance instance, SessionToken session);
}
\ No newline at end of file
Modified: branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstanceImpl.java
===================================================================
--- branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstanceImpl.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/main/java/org/teiid/net/socket/SocketServerInstanceImpl.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -30,7 +30,6 @@
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
-import java.net.SocketAddress;
import java.net.SocketTimeoutException;
import java.util.Iterator;
import java.util.Map;
@@ -53,7 +52,6 @@
import org.teiid.core.crypto.Cryptor;
import org.teiid.core.crypto.DhKeyGenerator;
import org.teiid.core.crypto.NullCryptor;
-import org.teiid.core.util.ApplicationInfo;
import org.teiid.net.CommunicationException;
import org.teiid.net.HostInfo;
import org.teiid.net.NetPlugin;
@@ -69,31 +67,28 @@
static final int HANDSHAKE_RETRIES = 10;
private static Logger log = Logger.getLogger("org.teiid.client.sockets"); //$NON-NLS-1$
- private AtomicInteger MESSAGE_ID = new AtomicInteger();
+ private static AtomicInteger MESSAGE_ID = new AtomicInteger();
private Map<Serializable, ResultsReceiver<Object>> asynchronousListeners = new ConcurrentHashMap<Serializable, ResultsReceiver<Object>>();
- private HostInfo hostInfo;
- private boolean ssl;
private long synchTimeout;
+ private HostInfo info;
private ObjectChannel socketChannel;
private Cryptor cryptor;
+ private String serverVersion;
private boolean hasReader;
- public SocketServerInstanceImpl() {
-
- }
-
- public SocketServerInstanceImpl(final HostInfo host, boolean ssl, long synchTimeout) {
- this.hostInfo = host;
- this.ssl = ssl;
+ public SocketServerInstanceImpl(HostInfo info, long synchTimeout) {
+ if (!info.isResolved()) {
+ throw new AssertionError("Expected HostInfo to be resolved"); //$NON-NLS-1$
+ }
+ this.info = info;
this.synchTimeout = synchTimeout;
}
public synchronized void connect(ObjectChannelFactory channelFactory) throws CommunicationException, IOException {
- InetSocketAddress address = new InetSocketAddress(hostInfo.getInetAddress(), hostInfo.getPortNumber());
- this.socketChannel = channelFactory.createObjectChannel(address, ssl);
+ this.socketChannel = channelFactory.createObjectChannel(new InetSocketAddress(info.getInetAddress(), info.getPortNumber()), info.isSsl());
try {
doHandshake();
} catch (CommunicationException e) {
@@ -105,20 +100,11 @@
}
}
- @Override
+ @Override
public HostInfo getHostInfo() {
- return this.hostInfo;
+ return info;
}
- @Override
- public SocketAddress getRemoteAddress() {
- return this.socketChannel.getRemoteAddress();
- }
-
- static String getVersionInfo() {
- return ApplicationInfo.getInstance().getMajorReleaseNumber();
- }
-
private void doHandshake() throws IOException, CommunicationException {
Handshake handshake = null;
for (int i = 0; i < HANDSHAKE_RETRIES; i++) {
@@ -143,9 +129,9 @@
/*if (!getVersionInfo().equals(handshake.getVersion())) {
throw new CommunicationException(NetPlugin.Util.getString("SocketServerInstanceImpl.version_mismatch", getVersionInfo(), handshake.getVersion())); //$NON-NLS-1$
}*/
+ serverVersion = handshake.getVersion();
+ handshake.setVersion();
- handshake.setVersion(getVersionInfo());
-
byte[] serverPublicKey = handshake.getPublicKey();
if (serverPublicKey != null) {
@@ -162,12 +148,17 @@
throw new CommunicationException(err);
}
}
+
+ @Override
+ public String getServerVersion() {
+ return serverVersion;
+ }
public boolean isOpen() {
return socketChannel.isOpen();
}
- protected void send(Message message, ResultsReceiver<Object> listener, Serializable messageKey)
+ public void send(Message message, ResultsReceiver<Object> listener, Serializable messageKey)
throws CommunicationException, InterruptedException {
if (listener != null) {
asynchronousListeners.put(messageKey, listener);
@@ -227,6 +218,7 @@
listener.receiveResults(messagePacket.getContents());
}
} else {
+ //TODO: could ping back
log.log(Level.FINE, "packet ignored:" + packet); //$NON-NLS-1$
}
}
@@ -242,7 +234,7 @@
return this.cryptor;
}
- void read(long timeout, TimeUnit unit, ResultsFuture<?> future) throws TimeoutException, InterruptedException {
+ public void read(long timeout, TimeUnit unit, ResultsFuture<?> future) throws TimeoutException, InterruptedException {
long timeoutMillis = (int)Math.min(unit.toMillis(timeout), Integer.MAX_VALUE);
long start = System.currentTimeMillis();
while (!future.isDone()) {
@@ -280,15 +272,24 @@
}
}
}
-
+
@SuppressWarnings("unchecked")
@Override
public <T> T getService(Class<T> iface) {
- return (T)Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {iface}, new RemoteInvocationHandler(iface));
+ return (T)Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {iface}, new RemoteInvocationHandler(iface) {
+ @Override
+ protected SocketServerInstanceImpl getInstance() {
+ return SocketServerInstanceImpl.this;
+ }
+ });
}
- public class RemoteInvocationHandler implements InvocationHandler {
+ public long getSynchTimeout() {
+ return synchTimeout;
+ }
+ public static abstract class RemoteInvocationHandler implements InvocationHandler {
+
private boolean secure;
private Class<?> targetClass;
@@ -302,17 +303,18 @@
throws Throwable {
Throwable t = null;
try {
+ final SocketServerInstance instance = getInstance();
Message message = new Message();
message.setContents(new ServiceInvocationStruct(args, method.getName(),
targetClass));
if (secure) {
- message.setContents(getCryptor().sealObject(message.getContents()));
+ message.setContents(instance.getCryptor().sealObject(message.getContents()));
}
ResultsFuture<Object> results = new ResultsFuture<Object>() {
@Override
protected Object convertResult() throws ExecutionException {
try {
- Object result = getCryptor().unsealObject((Serializable) super.convertResult());
+ Object result = instance.getCryptor().unsealObject(super.convertResult());
if (result instanceof ExceptionHolder) {
throw new ExecutionException(((ExceptionHolder)result).getException());
}
@@ -342,17 +344,17 @@
public Object get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException,
TimeoutException {
- read(timeout, unit, this);
+ instance.read(timeout, unit, this);
return super.get(timeout, unit);
}
};
final ResultsReceiver<Object> receiver = results.getResultsReceiver();
- send(message, receiver, Integer.valueOf(MESSAGE_ID.getAndIncrement()));
+ instance.send(message, receiver, Integer.valueOf(MESSAGE_ID.getAndIncrement()));
if (ResultsFuture.class.isAssignableFrom(method.getReturnType())) {
return results;
}
- return results.get(synchTimeout, TimeUnit.MILLISECONDS);
+ return results.get(instance.getSynchTimeout(), TimeUnit.MILLISECONDS);
} catch (ExecutionException e) {
t = e.getCause();
} catch (TimeoutException e) {
@@ -362,6 +364,8 @@
}
throw ExceptionUtil.convertException(method, t);
}
+
+ protected abstract SocketServerInstance getInstance() throws CommunicationException;
}
Modified: branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestAdminApiServerDiscovery.java
===================================================================
--- branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestAdminApiServerDiscovery.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestAdminApiServerDiscovery.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -35,7 +35,6 @@
import org.teiid.client.security.LogonResult;
import org.teiid.net.HostInfo;
import org.teiid.net.TeiidURL;
-import org.teiid.net.socket.SocketServerInstance;
public class TestAdminApiServerDiscovery extends TestCase {
@@ -65,7 +64,7 @@
processes.add(p2);
Mockito.stub(serverAdmin.getProcesses("*")).toReturn(processes); //$NON-NLS-1$
Mockito.stub(instance.getService(Admin.class)).toReturn(serverAdmin);
- Mockito.stub(instance.getHostInfo()).toReturn(knownHost);
+ Mockito.stub(instance.getHostInfo().getHostName()).toReturn("foo"); //$NON-NLS-1$
discovery.connectionSuccessful(knownHost);
List<HostInfo> knownHosts = discovery.getKnownHosts(new LogonResult(), instance);
Modified: branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestSocketServerConnection.java
===================================================================
--- branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestSocketServerConnection.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestSocketServerConnection.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -25,29 +25,30 @@
*/
package org.teiid.net.socket;
+import static org.junit.Assert.*;
+
import java.io.IOException;
-import java.net.InetSocketAddress;
+import java.io.Serializable;
+import java.util.Collection;
import java.util.Properties;
-import junit.framework.TestCase;
-
+import org.junit.Test;
import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
import org.teiid.client.security.ILogon;
import org.teiid.client.security.InvalidSessionException;
import org.teiid.client.security.LogonException;
import org.teiid.client.security.LogonResult;
import org.teiid.client.security.SessionToken;
import org.teiid.client.util.ResultsFuture;
+import org.teiid.client.util.ResultsReceiver;
import org.teiid.core.TeiidComponentException;
import org.teiid.core.crypto.NullCryptor;
import org.teiid.net.CommunicationException;
import org.teiid.net.ConnectionException;
import org.teiid.net.HostInfo;
import org.teiid.net.TeiidURL;
-import org.teiid.net.socket.SocketServerConnection;
-import org.teiid.net.socket.SocketServerConnectionFactory;
-import org.teiid.net.socket.SocketServerInstance;
-import org.teiid.net.socket.SocketServerInstanceFactory;
/**
@@ -55,12 +56,16 @@
* @see SocketServerConnection
* @since Westport
*/
-public class TestSocketServerConnection extends TestCase {
+public class TestSocketServerConnection {
private static final class FakeILogon implements ILogon {
Throwable t;
+ public FakeILogon(Throwable t) {
+ this.t = t;
+ }
+
@Override
public void assertIdentity(SessionToken sessionId)
throws InvalidSessionException, TeiidComponentException {
@@ -83,15 +88,25 @@
@Override
public ResultsFuture<?> ping()
- throws InvalidSessionException,
- TeiidComponentException {
+ throws TeiidComponentException, CommunicationException {
if (t != null) {
+ if (t instanceof CommunicationException) {
+ CommunicationException ce = (CommunicationException)t;
+ t = null;
+ throw ce;
+ }
TeiidComponentException e = new TeiidComponentException(t);
t = null;
throw e;
}
- return null;
+ return ResultsFuture.NULL_FUTURE;
}
+
+ @Override
+ public ResultsFuture<?> ping(Collection<String> sessions)
+ throws TeiidComponentException, CommunicationException {
+ return ping();
+ }
}
/**
@@ -109,7 +124,7 @@
*
* @since Westport
*/
- public void testSocketServerConnection_PropertiesClientHost() throws Throwable {
+ @Test public void testSocketServerConnection_PropertiesClientHost() throws Throwable {
Properties p = new Properties();
SocketServerConnectionFactory.updateConnectionProperties(p);
@@ -118,25 +133,20 @@
assertTrue(p.containsKey(TeiidURL.CONNECTION.CLIENT_IP_ADDRESS));
}
- public void testLogonFailsWithMultipleHosts() throws Exception {
+ @Test public void testLogonFailsWithMultipleHosts() throws Exception {
Properties p = new Properties();
- SocketServerInstanceFactory instanceFactory = new SocketServerInstanceFactory() {
- @Override
- public SocketServerInstance getServerInstance(HostInfo info,
- boolean ssl) throws CommunicationException, IOException {
- throw new SingleInstanceCommunicationException();
- }
- };
+ SocketServerInstanceFactory instanceFactory = Mockito.mock(SocketServerInstanceFactory.class);
+ Mockito.stub(instanceFactory.getServerInstance((HostInfo)Mockito.anyObject())).toThrow(new SingleInstanceCommunicationException());
ServerDiscovery discovery = new UrlServerDiscovery(new TeiidURL("mm://host1:1,host2:2")); //$NON-NLS-1$
try {
- new SocketServerConnection(instanceFactory, false, discovery, p, null);
+ new SocketServerConnection(instanceFactory, false, discovery, p);
fail("exception expected"); //$NON-NLS-1$
} catch (CommunicationException e) {
assertEquals("No valid host available. Attempted connections to: [host1:1, host2:2]", e.getMessage()); //$NON-NLS-1$
}
}
- public void testLogon() throws Exception {
+ @Test public void testLogon() throws Exception {
SocketServerConnection connection = createConnection(null);
assertEquals(String.valueOf(1), connection.getLogonResult().getSessionID());
}
@@ -144,63 +154,83 @@
/**
* Since the original instance is still open, this will be a transparent retry
*/
- public void testRetry() throws Exception {
+ @Test public void testRetry() throws Exception {
SocketServerConnection connection = createConnection(new SingleInstanceCommunicationException());
connection.setFailOver(true);
+ connection.setFailOverPingInterval(50);
ILogon logon = connection.getService(ILogon.class);
+ Thread.sleep(70);
logon.ping();
}
- public void testImmediateFail() throws Exception {
+ @Test(expected=CommunicationException.class) public void testImmediateFail() throws Exception {
SocketServerConnection connection = createConnection(new CommunicationException());
ILogon logon = connection.getService(ILogon.class);
- try {
- logon.ping();
- fail("expected exception"); //$NON-NLS-1$
- } catch (TeiidComponentException e) {
-
- }
+ logon.ping();
}
- public void testImmediateFail1() throws Exception {
+ @Test(expected=CommunicationException.class) public void testImmediateFail1() throws Exception {
SocketServerConnection connection = createConnection(new CommunicationException());
connection.setFailOver(true);
ILogon logon = connection.getService(ILogon.class);
- try {
- logon.ping();
- fail("expected exception"); //$NON-NLS-1$
- } catch (TeiidComponentException e) {
-
- }
+ logon.ping();
}
private SocketServerConnection createConnection(final Throwable throwException) throws CommunicationException, ConnectionException {
return createConnection(throwException, new HostInfo("0.0.0.2", 1)); //$NON-NLS-1$
}
- private SocketServerConnection createConnection(final Throwable t, HostInfo hostInfo)
+ private SocketServerConnection createConnection(final Throwable t, final HostInfo hostInfo)
throws CommunicationException, ConnectionException {
Properties p = new Properties();
ServerDiscovery discovery = new UrlServerDiscovery(new TeiidURL(hostInfo.getHostName(), hostInfo.getPortNumber(), false));
SocketServerInstanceFactory instanceFactory = new SocketServerInstanceFactory() {
+ FakeILogon logon = new FakeILogon(t);
+
@Override
- public SocketServerInstance getServerInstance(final HostInfo info,
- boolean ssl) throws CommunicationException, IOException {
+ public SocketServerInstance getServerInstance(HostInfo info)
+ throws CommunicationException, IOException {
SocketServerInstance instance = Mockito.mock(SocketServerInstance.class);
Mockito.stub(instance.getCryptor()).toReturn(new NullCryptor());
- Mockito.stub(instance.getRemoteAddress()).toReturn(new InetSocketAddress(info.getInetAddress(), info.getPortNumber()));
- FakeILogon logon = new FakeILogon();
- logon.t = t;
+ Mockito.stub(instance.getHostInfo()).toReturn(hostInfo);
Mockito.stub(instance.getService(ILogon.class)).toReturn(logon);
+ if (t != null) {
+ try {
+ Mockito.doAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation)
+ throws Throwable {
+ if (logon.t == null) {
+ return null;
+ }
+ throw logon.t;
+ }
+ }).when(instance).send((Message)Mockito.anyObject(), (ResultsReceiver<Object>)Mockito.anyObject(), (Serializable)Mockito.anyObject());
+ } catch (Exception e) {
+
+ }
+ }
Mockito.stub(instance.isOpen()).toReturn(true);
return instance;
}
+
+ @Override
+ public void connected(SocketServerInstance instance,
+ SessionToken session) {
+
+ }
+
+ @Override
+ public void disconnected(SocketServerInstance instance,
+ SessionToken session) {
+
+ }
};
- SocketServerConnection connection = new SocketServerConnection(instanceFactory, false, discovery, p, null);
+ SocketServerConnection connection = new SocketServerConnection(instanceFactory, false, discovery, p);
return connection;
}
- public void testIsSameInstance() throws Exception {
+ @Test public void testIsSameInstance() throws Exception {
SocketServerConnection conn = createConnection(null, new HostInfo("0.0.0.0", 1)); //$NON-NLS-1$
SocketServerConnection conn1 = createConnection(null, new HostInfo("0.0.0.1", 1)); //$NON-NLS-1$
Modified: branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestSocketServerInstanceImpl.java
===================================================================
--- branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestSocketServerInstanceImpl.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/client/src/test/java/org/teiid/net/socket/TestSocketServerInstanceImpl.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -34,17 +34,14 @@
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
-import org.junit.Ignore;
import org.junit.Test;
import org.teiid.client.security.ILogon;
import org.teiid.client.util.ResultsFuture;
import org.teiid.core.TeiidComponentException;
import org.teiid.net.CommunicationException;
import org.teiid.net.HostInfo;
-import org.teiid.net.socket.ObjectChannelFactory;
-import org.teiid.net.socket.SocketServerInstanceImpl;
-
+ at SuppressWarnings("nls")
public class TestSocketServerInstanceImpl {
private static class FakeObjectChannel implements ObjectChannel, ObjectChannelFactory {
@@ -127,7 +124,8 @@
private SocketServerInstanceImpl createInstance(ObjectChannelFactory channelFactory)
throws CommunicationException, IOException {
- SocketServerInstanceImpl ssii = new SocketServerInstanceImpl(new HostInfo("0.0.0.0", 1), false, 1); //$NON-NLS-1$
+ HostInfo info = new HostInfo("0.0.0.0", 1);
+ SocketServerInstanceImpl ssii = new SocketServerInstanceImpl(info, 1);
ssii.connect(channelFactory);
return ssii;
}
@@ -147,17 +145,4 @@
}
}
- @Ignore
- @Test public void testVersionMismatch() throws Exception {
- Handshake h = new Handshake();
- h.setVersion("foo"); //$NON-NLS-1$
- final FakeObjectChannel channel = new FakeObjectChannel(Arrays.asList(h));
- try {
- createInstance(channel);
- fail("exception expected"); //$NON-NLS-1$
- } catch (CommunicationException e) {
- assertTrue(e.getMessage().startsWith("Handshake failed due to version mismatch")); //$NON-NLS-1$
- }
- }
-
}
Modified: branches/7.1.x/runtime/src/main/java/org/teiid/transport/LocalServerConnection.java
===================================================================
--- branches/7.1.x/runtime/src/main/java/org/teiid/transport/LocalServerConnection.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/runtime/src/main/java/org/teiid/transport/LocalServerConnection.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -123,11 +123,11 @@
}
@Override
- public ResultsFuture<?> isOpen() {
+ public boolean isOpen(long msToTest) {
if (shutdown) {
- return null;
+ return false;
}
- return ResultsFuture.NULL_FUTURE;
+ return true;
}
public void close() {
Modified: branches/7.1.x/runtime/src/main/java/org/teiid/transport/LogonImpl.java
===================================================================
--- branches/7.1.x/runtime/src/main/java/org/teiid/transport/LogonImpl.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/runtime/src/main/java/org/teiid/transport/LogonImpl.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -22,6 +22,7 @@
package org.teiid.transport;
+import java.util.Collection;
import java.util.Properties;
import javax.security.auth.login.LoginException;
@@ -41,6 +42,7 @@
import org.teiid.dqp.service.SessionServiceException;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
+import org.teiid.net.CommunicationException;
import org.teiid.net.TeiidURL;
import org.teiid.security.Credentials;
@@ -107,6 +109,18 @@
LogManager.logTrace(LogConstants.CTX_SECURITY, "Ping", id); //$NON-NLS-1$
return ResultsFuture.NULL_FUTURE;
}
+
+ @Override
+ public ResultsFuture<?> ping(Collection<String> sessions)
+ throws TeiidComponentException, CommunicationException {
+ for (String string : sessions) {
+ try {
+ this.service.pingServer(string);
+ } catch (InvalidSessionException e) {
+ }
+ }
+ return ResultsFuture.NULL_FUTURE;
+ }
@Override
public void assertIdentity(SessionToken checkSession) throws InvalidSessionException, TeiidComponentException {
Modified: branches/7.1.x/runtime/src/main/java/org/teiid/transport/PgBackendProtocol.java
===================================================================
--- branches/7.1.x/runtime/src/main/java/org/teiid/transport/PgBackendProtocol.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/runtime/src/main/java/org/teiid/transport/PgBackendProtocol.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -39,7 +39,6 @@
import org.jboss.netty.channel.ChannelDownstreamHandler;
import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.channel.ChannelPipelineCoverage;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.MessageEvent;
import org.teiid.core.util.ReflectionHelper;
@@ -54,7 +53,6 @@
* Some parts of this code is taken from H2's implementation of ODBC
*/
@SuppressWarnings("nls")
- at ChannelPipelineCoverage("one")
public class PgBackendProtocol implements ChannelDownstreamHandler, ODBCClientRemote {
private static final int PG_TYPE_VARCHAR = 1043;
Modified: branches/7.1.x/runtime/src/main/java/org/teiid/transport/PgFrontendProtocol.java
===================================================================
--- branches/7.1.x/runtime/src/main/java/org/teiid/transport/PgFrontendProtocol.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/runtime/src/main/java/org/teiid/transport/PgFrontendProtocol.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -34,7 +34,6 @@
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.channel.ChannelPipelineCoverage;
import org.jboss.netty.handler.codec.frame.FrameDecoder;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
@@ -46,7 +45,6 @@
* Some parts of this code is taken from H2's implementation of ODBC
*/
@SuppressWarnings("nls")
- at ChannelPipelineCoverage("one")
public class PgFrontendProtocol extends FrameDecoder {
private int maxObjectSize;
Modified: branches/7.1.x/runtime/src/main/java/org/teiid/transport/SSLAwareChannelHandler.java
===================================================================
--- branches/7.1.x/runtime/src/main/java/org/teiid/transport/SSLAwareChannelHandler.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/runtime/src/main/java/org/teiid/transport/SSLAwareChannelHandler.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -42,13 +42,13 @@
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
-import org.jboss.netty.channel.ChannelPipelineCoverage;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.DefaultChannelPipeline;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
+import org.jboss.netty.channel.ChannelHandler.Sharable;
import org.jboss.netty.handler.ssl.SslHandler;
import org.jboss.netty.handler.stream.ChunkedWriteHandler;
import org.teiid.common.buffer.StorageManager;
@@ -62,7 +62,7 @@
* Main class for creating Netty Nio Channels
*/
- at ChannelPipelineCoverage(ChannelPipelineCoverage.ALL)
+ at Sharable
public class SSLAwareChannelHandler extends SimpleChannelHandler implements ChannelPipelineFactory {
public class ObjectChannelImpl implements ObjectChannel {
Modified: branches/7.1.x/runtime/src/main/java/org/teiid/transport/SocketClientInstance.java
===================================================================
--- branches/7.1.x/runtime/src/main/java/org/teiid/transport/SocketClientInstance.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/runtime/src/main/java/org/teiid/transport/SocketClientInstance.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -94,7 +94,6 @@
public void onConnection() throws CommunicationException {
Handshake handshake = new Handshake();
- handshake.setVersion(SocketListener.getVersionInfo());
if (usingEncryption) {
keyGen = new DhKeyGenerator();
Modified: branches/7.1.x/runtime/src/main/java/org/teiid/transport/SocketListener.java
===================================================================
--- branches/7.1.x/runtime/src/main/java/org/teiid/transport/SocketListener.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/runtime/src/main/java/org/teiid/transport/SocketListener.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -31,7 +31,6 @@
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.teiid.common.buffer.StorageManager;
-import org.teiid.core.util.ApplicationInfo;
import org.teiid.core.util.NamedThreadFactory;
import org.teiid.logging.LogConstants;
import org.teiid.logging.LogManager;
@@ -99,10 +98,6 @@
return ((InetSocketAddress)this.serverChanel.getLocalAddress()).getPort();
}
- static String getVersionInfo() {
- return ApplicationInfo.getInstance().getMajorReleaseNumber();
- }
-
public void stop() {
this.serverChanel.close();
this.nettyPool.shutdownNow();
Modified: branches/7.1.x/runtime/src/test/java/org/teiid/transport/TestCommSockets.java
===================================================================
--- branches/7.1.x/runtime/src/test/java/org/teiid/transport/TestCommSockets.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/runtime/src/test/java/org/teiid/transport/TestCommSockets.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -37,6 +37,7 @@
import org.teiid.client.security.ILogon;
import org.teiid.client.security.LogonException;
import org.teiid.client.security.LogonResult;
+import org.teiid.client.security.SessionToken;
import org.teiid.common.buffer.BufferManagerFactory;
import org.teiid.core.ComponentNotFoundException;
import org.teiid.core.crypto.NullCryptor;
@@ -69,19 +70,14 @@
}
}
- @Test public void testFailedConnect() throws Exception {
+ @Test(expected=CommunicationException.class) public void testFailedConnect() throws Exception {
SSLConfiguration config = new SSLConfiguration();
listener = new SocketListener(addr.getPort(), addr.getAddress().getHostAddress(),1024, 1024, 1, config, null, BufferManagerFactory.getStandaloneBufferManager());
- try {
- Properties p = new Properties();
- String url = new TeiidURL(addr.getHostName(), listener.getPort() - 1, false).getAppServerURL();
- p.setProperty(TeiidURL.CONNECTION.SERVER_URL, url); //wrong port
- SocketServerConnectionFactory.getInstance().getConnection(p);
- fail("exception expected"); //$NON-NLS-1$
- } catch (CommunicationException e) {
-
- }
+ Properties p = new Properties();
+ String url = new TeiidURL(addr.getHostName(), listener.getPort() - 1, false).getAppServerURL();
+ p.setProperty(TeiidURL.CONNECTION.SERVER_URL, url); //wrong port
+ SocketServerConnectionFactory.getInstance().getConnection(p);
}
@Test public void testConnectWithoutPooling() throws Exception {
@@ -157,7 +153,7 @@
@Override
public LogonResult logon(Properties connProps)
throws LogonException, ComponentNotFoundException {
- return new LogonResult();
+ return new LogonResult(new SessionToken("dummy"), "x", 1, "z");
}
}, null);
Modified: branches/7.1.x/runtime/src/test/java/org/teiid/transport/TestSocketRemoting.java
===================================================================
--- branches/7.1.x/runtime/src/test/java/org/teiid/transport/TestSocketRemoting.java 2010-08-19 15:30:38 UTC (rev 2481)
+++ branches/7.1.x/runtime/src/test/java/org/teiid/transport/TestSocketRemoting.java 2010-08-20 22:18:46 UTC (rev 2482)
@@ -29,6 +29,10 @@
import java.io.Reader;
import java.io.Serializable;
import java.io.StringReader;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.Collection;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -60,7 +64,7 @@
import org.teiid.net.socket.SocketServerInstanceImpl;
import org.teiid.net.socket.UrlServerDiscovery;
-
+ at SuppressWarnings("nls")
public class TestSocketRemoting {
public interface FakeService {
@@ -104,15 +108,11 @@
ClientServiceRegistryImpl server;
private ResultsReceiver<Object> listener;
- public FakeClientServerInstance(ClientServiceRegistryImpl server) {
- super();
+ public FakeClientServerInstance(ClientServiceRegistryImpl server) throws UnknownHostException {
+ super(new HostInfo("foo", new InetSocketAddress(InetAddress.getLocalHost(), 1)), 1000);
this.server = server;
}
-
- public HostInfo getHostInfo() {
- return new HostInfo("fake", 1); //$NON-NLS-1$
- }
-
+
public boolean isOpen() {
return true;
}
@@ -182,6 +182,12 @@
}
@Override
+ public ResultsFuture<?> ping(Collection<String> sessions)
+ throws TeiidComponentException, CommunicationException {
+ return null;
+ }
+
+ @Override
public void assertIdentity(SessionToken sessionId)
throws InvalidSessionException,
TeiidComponentException {
@@ -226,12 +232,24 @@
SocketServerConnection connection = new SocketServerConnection(new SocketServerInstanceFactory() {
@Override
- public SocketServerInstance getServerInstance(HostInfo info,
- boolean ssl) throws CommunicationException, IOException {
+ public SocketServerInstance getServerInstance(HostInfo info)
+ throws CommunicationException, IOException {
return serverInstance;
}
- }, false, new UrlServerDiscovery(new TeiidURL("foo", 1, false)), new Properties(), null); //$NON-NLS-1$
+ @Override
+ public void connected(SocketServerInstance instance,
+ SessionToken session) {
+
+ }
+
+ @Override
+ public void disconnected(SocketServerInstance instance,
+ SessionToken session) {
+
+ }
+
+ }, false, new UrlServerDiscovery(new TeiidURL("0.0.0.0", 1, false)), new Properties()); //$NON-NLS-1$
return connection;
}
More information about the teiid-commits
mailing list