You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tomee.apache.org by db...@apache.org on 2008/09/09 02:09:27 UTC
svn commit: r693321 [1/2] - in /openejb/trunk/openejb3:
container/openejb-core/src/main/java/org/apache/openejb/util/
server/openejb-client/src/main/java/org/apache/openejb/client/
server/openejb-client/src/test/java/org/apache/openejb/client/ server/o...
Author: dblevins
Date: Mon Sep 8 17:09:25 2008
New Revision: 693321
URL: http://svn.apache.org/viewvc?rev=693321&view=rev
Log:
OPENEJB-903: Multicast discovery and grouping
OPENEJB-911: Graceful shutdown of client/server connections
OPENEJB-857: Client connection KeepAlive (had to rewrite this for OPENEJB-911)
Added a ClusterMetaData which is similar to what the ServerMetaData was aiming to be. The issue with the ServerMetaData is that it's tracked on a per-ejb-proxy basis and any updates to the list of servers in the cluster are only reflected in the proxy immediately used. All other proxies will still hold onto the outdated list. Second, not all request types could be clustered and have failover, essentially only ejb requests could failover, jndi and authentication could not. Now the ClusterMetaData version associated with the ServerMetaData is sent to the server *before* the main request and then the server can send back a new list regardless of which type of request it is.
Additionally code has been added to guarantee that the ServerMetaData the client used to connet with the server or cluster is the used in an EJB proxies created via the IntraVM server when the are replaced for Remote Server proxies at serialization. Previously they were sent back with "foo://localhost:4201" as the server address which was not good.
Added:
openejb/trunk/openejb3/container/openejb-core/src/main/java/org/apache/openejb/util/Exceptions.java
openejb/trunk/openejb3/container/openejb-core/src/main/java/org/apache/openejb/util/VersionedSet.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterMetaData.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterRequest.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterResponse.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/KeepAliveStyle.java
openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/ClusterRequestHandler.java
openejb/trunk/openejb3/server/openejb-ejbd/src/test/java/org/apache/openejb/server/ejbd/FailoverTest.java
openejb/trunk/openejb3/server/openejb-server/src/main/java/org/apache/openejb/server/DiscoveryAgent.java
- copied, changed from r691472, openejb/trunk/openejb3/server/openejb-discovery/src/main/java/org/apache/openejb/server/discovery/DiscoveryAgent.java
openejb/trunk/openejb3/server/openejb-server/src/main/java/org/apache/openejb/server/DiscoveryListener.java
- copied, changed from r691472, openejb/trunk/openejb3/server/openejb-discovery/src/main/java/org/apache/openejb/server/discovery/DiscoveryListener.java
openejb/trunk/openejb3/server/openejb-server/src/main/java/org/apache/openejb/server/DiscoveryRegistry.java
openejb/trunk/openejb3/server/openejb-server/src/main/java/org/apache/openejb/server/ServerServiceFilter.java
Removed:
openejb/trunk/openejb3/server/openejb-discovery/src/main/java/org/apache/openejb/server/discovery/DiscoveryAgent.java
openejb/trunk/openejb3/server/openejb-discovery/src/main/java/org/apache/openejb/server/discovery/DiscoveryListener.java
Modified:
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/Client.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ConnectionManager.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ConnectionStrategy.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/EJBRequest.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ProtocolMetaData.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/RequestMethodConstants.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ServerMetaData.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/SocketConnectionFactory.java
openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/StickyConnectionStrategy.java
openejb/trunk/openejb3/server/openejb-client/src/test/java/org/apache/openejb/client/StickyConnectionStrategyTest.java
openejb/trunk/openejb3/server/openejb-discovery/src/main/java/org/apache/openejb/server/discovery/MulticastDiscoveryAgent.java
openejb/trunk/openejb3/server/openejb-discovery/src/test/java/org/apache/openejb/server/discovery/MulticastDiscoveryAgentTest.java
openejb/trunk/openejb3/server/openejb-ejbd/pom.xml
openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/CallContext.java
openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/ClientObjectFactory.java
openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbDaemon.java
openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbServer.java
openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/KeepAliveServer.java
openejb/trunk/openejb3/server/openejb-server/src/main/java/org/apache/openejb/server/ServiceDaemon.java
openejb/trunk/openejb3/server/openejb-server/src/main/java/org/apache/openejb/server/ServiceManager.java
openejb/trunk/openejb3/server/openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java
Added: openejb/trunk/openejb3/container/openejb-core/src/main/java/org/apache/openejb/util/Exceptions.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/container/openejb-core/src/main/java/org/apache/openejb/util/Exceptions.java?rev=693321&view=auto
==============================================================================
--- openejb/trunk/openejb3/container/openejb-core/src/main/java/org/apache/openejb/util/Exceptions.java (added)
+++ openejb/trunk/openejb3/container/openejb-core/src/main/java/org/apache/openejb/util/Exceptions.java Mon Sep 8 17:09:25 2008
@@ -0,0 +1,96 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.util;
+
+import javax.ejb.EJBException;
+import javax.naming.AuthenticationException;
+import javax.naming.NamingException;
+import javax.transaction.RollbackException;
+import java.io.IOException;
+import java.io.NotSerializableException;
+import java.rmi.RemoteException;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class Exceptions {
+
+ /**
+ * Removes the need for a cast when using initCause
+ * @param t
+ * @param cause
+ * @return
+ */
+ public static <T extends Throwable> T initCause(T t, Throwable cause) {
+ return (T) t.initCause(cause);
+ }
+
+
+ public static IOException newIOException(String message, Throwable cause){
+ return initCause(new IOException(message), cause);
+ }
+
+ public static IOException newIOException(Throwable cause){
+ return initCause(new IOException(), cause);
+ }
+
+ public static NamingException newNamingException(String message, Throwable cause){
+ return initCause(new NamingException(message), cause);
+ }
+
+ public static NamingException newNamingException(Throwable cause){
+ return initCause(new NamingException(), cause);
+ }
+
+
+ public static RollbackException newRollbackException(String message, Throwable cause){
+ return initCause(new RollbackException(message), cause);
+ }
+
+ public static RollbackException newRollbackException(Throwable cause){
+ return initCause(new RollbackException(), cause);
+ }
+
+
+ public static AuthenticationException newAuthenticationException(String message, Throwable cause){
+ return initCause(new AuthenticationException(message), cause);
+ }
+
+ public static AuthenticationException newAuthenticationException(Throwable cause){
+ return initCause(new AuthenticationException(), cause);
+ }
+
+
+ public static EJBException newEJBException(String message, Throwable cause){
+ return initCause(new EJBException(message), cause);
+ }
+
+ public static EJBException newEJBException(Throwable cause){
+ return initCause(new EJBException(), cause);
+ }
+
+
+ public static NotSerializableException newNotSerializableException(String message, Throwable cause){
+ return initCause(new NotSerializableException(message), cause);
+ }
+
+ public static NotSerializableException newNotSerializableException(Throwable cause){
+ return initCause(new NotSerializableException(), cause);
+ }
+
+
+}
Added: openejb/trunk/openejb3/container/openejb-core/src/main/java/org/apache/openejb/util/VersionedSet.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/container/openejb-core/src/main/java/org/apache/openejb/util/VersionedSet.java?rev=693321&view=auto
==============================================================================
--- openejb/trunk/openejb3/container/openejb-core/src/main/java/org/apache/openejb/util/VersionedSet.java (added)
+++ openejb/trunk/openejb3/container/openejb-core/src/main/java/org/apache/openejb/util/VersionedSet.java Mon Sep 8 17:09:25 2008
@@ -0,0 +1,161 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.util;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * @version $Revision$ $Date$
+ */
+public class VersionedSet<E> {
+
+ private Set<E> current;
+
+ private ReadWriteLock sync = new ReentrantReadWriteLock();
+
+ public boolean add(E o) {
+ Lock lock = sync.writeLock();
+ lock.lock();
+ Set nextVersion = null;
+ try {
+ if (!current.set().contains(o)) {
+ nextVersion = new Set(current);
+ return nextVersion.set().add(o);
+ } else {
+ return false;
+ }
+ } finally {
+ if (nextVersion != null) current = nextVersion;
+ lock.unlock();
+ }
+ }
+
+ public boolean remove(Object o) {
+ Lock lock = sync.writeLock();
+ lock.lock();
+ Set nextVersion = null;
+ try {
+ if (current.set().contains(o)) {
+ nextVersion = new Set(current);
+ return nextVersion.set().remove(o);
+ } else {
+ return false;
+ }
+ } finally {
+ if (nextVersion != null) current = nextVersion;
+ lock.unlock();
+ }
+ }
+
+ public Set<E> currentSet() {
+ Lock lock = sync.readLock();
+ lock.lock();
+ try {
+ return current;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public static class Set<E> implements java.util.Set<E> {
+ private final int version;
+ private final java.util.Set set;
+
+ Set() {
+ version = 0;
+ set = new LinkedHashSet<E>();
+ }
+
+ Set(Set v) {
+ this.version = v.version + 1;
+ this.set = new LinkedHashSet<E>(v.set);
+ }
+
+ java.util.Set set() {
+ return set;
+ }
+
+ public int getVersion() {
+ return version;
+ }
+
+ public boolean add(Object o) {
+ throw new UnsupportedOperationException();
+ }
+
+ public boolean addAll(Collection c) {
+ throw new UnsupportedOperationException();
+ }
+
+ public void clear() {
+ throw new UnsupportedOperationException();
+ }
+
+ public boolean contains(Object o) {
+ return set.contains(o);
+ }
+
+ public boolean containsAll(Collection c) {
+ return set.containsAll(c);
+ }
+
+ public boolean equals(Object o) {
+ return set.equals(o);
+ }
+
+ public int hashCode() {
+ return set.hashCode();
+ }
+
+ public boolean isEmpty() {
+ return set.isEmpty();
+ }
+
+ public Iterator iterator() {
+ return set.iterator();
+ }
+
+ public boolean remove(Object o) {
+ throw new UnsupportedOperationException();
+ }
+
+ public boolean removeAll(Collection c) {
+ throw new UnsupportedOperationException();
+ }
+
+ public boolean retainAll(Collection c) {
+ throw new UnsupportedOperationException();
+ }
+
+ public int size() {
+ return set.size();
+ }
+
+ public Object[] toArray() {
+ return set.toArray();
+ }
+
+ public Object[] toArray(Object[] a) {
+ return set.toArray(a);
+ }
+ }
+}
Modified: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/Client.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/Client.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/Client.java (original)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/Client.java Mon Sep 8 17:09:25 2008
@@ -16,6 +16,7 @@
*/
package org.apache.openejb.client;
+import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInput;
@@ -23,13 +24,15 @@
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.rmi.RemoteException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Level;
import java.util.logging.Logger;
public class Client {
private static final Logger logger = Logger.getLogger("OpenEJB.client");
- private static final ProtocolMetaData PROTOCOL_VERSION = new ProtocolMetaData("3.0");
+ private static final ProtocolMetaData PROTOCOL_VERSION = new ProtocolMetaData("3.1");
private static Client client = new Client();
@@ -43,15 +46,19 @@
}
protected Response processRequest(Request req, Response res, ServerMetaData server) throws RemoteException {
+// System.out.println("req = " + req);
if (server == null)
throw new IllegalArgumentException("Server instance cannot be null");
Connection conn = null;
try {
+
+ ClusterMetaData cluster = getClusterMetaData(server);
+
/*----------------------------*/
/* Get a connection to server */
/*----------------------------*/
- conn = ConnectionManager.getConnection(server);
+ conn = ConnectionManager.getConnection(cluster, server);
/*----------------------------------*/
/* Get output streams */
@@ -71,41 +78,65 @@
/*----------------------------------*/
/* Write the protocol magic */
/*----------------------------------*/
- try{
+ try {
PROTOCOL_VERSION.writeExternal(out);
- } catch (Throwable e){
- throw new RemoteException("Cannot write the protocol metadata to the server: " , e );
+ } catch (Throwable e) {
+ throw new RemoteException("Cannot write the protocol metadata to the server: ", e);
}
/*----------------------------------*/
- /* Write request type */
+ /* Get output streams */
/*----------------------------------*/
+ ObjectOutput objectOut;
try {
- out.write(req.getRequestType());
+ objectOut = new ObjectOutputStream(out);
} catch (IOException e) {
- throw new RemoteException("Cannot write the request type to the server: ", e);
+ throw new RemoteException("Cannot open object output stream to server: ", e);
} catch (Throwable e) {
- throw new RemoteException("Cannot write the request type to the server: ", e);
+ throw new RemoteException("Cannot open object output stream to server: ", e);
}
/*----------------------------------*/
- /* Get output streams */
+ /* Write ServerMetaData */
/*----------------------------------*/
- ObjectOutput objectOut;
try {
- objectOut = new ObjectOutputStream(out);
+ server.writeExternal(objectOut);
+
+ } catch (Throwable e) {
+ throw new RemoteException("Cannot write the ServerMetaData to the server: ", e);
+ }
+
+ /*----------------------------------*/
+ /* Write ClusterMetaData */
+ /*----------------------------------*/
+ try {
+
+ ClusterRequest clusterRequest = new ClusterRequest(cluster);
+ objectOut.write(clusterRequest.getRequestType());
+ clusterRequest.writeExternal(objectOut);
+
+ } catch (Throwable e) {
+ throw new RemoteException("Cannot write the ServerMetaData to the server: ", e);
+ }
+
+ /*----------------------------------*/
+ /* Write request type */
+ /*----------------------------------*/
+ try {
+
+ objectOut.write(req.getRequestType());
} catch (IOException e) {
- throw new RemoteException("Cannot open object output stream to server: ", e);
+ throw new RemoteException("Cannot write the request type to the server: ", e);
} catch (Throwable e) {
- throw new RemoteException("Cannot open object output stream to server: ", e);
+ throw new RemoteException("Cannot write the request type to the server: ", e);
}
/*----------------------------------*/
@@ -136,8 +167,9 @@
in = conn.getInputStream();
+
} catch (IOException e) {
- throw new RemoteException("Cannot open input stream to server: " , e );
+ throw new RemoteException("Cannot open input stream to server: ", e);
}
ProtocolMetaData protocolMetaData = null;
@@ -146,17 +178,44 @@
protocolMetaData = new ProtocolMetaData();
protocolMetaData.readExternal(in);
+ } catch (EOFException e) {
+ throw new RemoteException("Prematurely reached the end of the stream. " + protocolMetaData.getSpec(), e);
} catch (IOException e) {
- throw new RemoteException("Cannot deternmine server protocol version: Received "+protocolMetaData.getSpec() , e );
+ throw new RemoteException("Cannot deternmine server protocol version: Received " + protocolMetaData.getSpec(), e);
}
ObjectInput objectIn;
- try{
+ try {
objectIn = new EjbObjectInputStream(in);
- } catch (Throwable e){
- throw new RemoteException("Cannot open object input stream to server ("+protocolMetaData.getSpec() +") : "+e.getMessage() , e );
+ } catch (Throwable e) {
+ throw new RemoteException("Cannot open object input stream to server (" + protocolMetaData.getSpec() + ") : " + e.getMessage(), e);
+ }
+
+ /*----------------------------------*/
+ /* Read response */
+ /*----------------------------------*/
+ try {
+ ClusterResponse clusterResponse = new ClusterResponse();
+ clusterResponse.readExternal(objectIn);
+ switch (clusterResponse.getResponseCode()) {
+ case UPDATE: {
+ clusters.put(server, clusterResponse.getUpdatedMetaData());
+ }
+ break;
+ case FAILURE: {
+ throw clusterResponse.getFailure();
+ }
+ }
+ } catch (ClassNotFoundException e) {
+ throw new RemoteException("Cannot read the response from the server. The class for an object being returned is not located in this system:", e);
+
+ } catch (IOException e) {
+ throw new RemoteException("Cannot read the response from the server (" + protocolMetaData.getSpec() + ") : " + e.getMessage(), e);
+
+ } catch (Throwable e) {
+ throw new RemoteException("Error reading response from server (" + protocolMetaData.getSpec() + ") : " + e.getMessage(), e);
}
/*----------------------------------*/
@@ -168,13 +227,13 @@
} catch (ClassNotFoundException e) {
throw new RemoteException("Cannot read the response from the server. The class for an object being returned is not located in this system:", e);
- } catch (IOException e){
- throw new RemoteException("Cannot read the response from the server ("+protocolMetaData.getSpec() +") : "+e.getMessage() , e );
+ } catch (IOException e) {
+ throw new RemoteException("Cannot read the response from the server (" + protocolMetaData.getSpec() + ") : " + e.getMessage(), e);
- } catch (Throwable e){
- throw new RemoteException("Error reading response from server ("+protocolMetaData.getSpec() +") : "+e.getMessage() , e );
+ } catch (Throwable e) {
+ throw new RemoteException("Error reading response from server (" + protocolMetaData.getSpec() + ") : " + e.getMessage(), e);
}
-
+
} catch (RemoteException e) {
throw e;
} catch (Throwable error) {
@@ -182,9 +241,9 @@
} finally {
try {
- if (conn != null) {
- conn.close();
- }
+ if (conn != null) {
+ conn.close();
+ }
} catch (Throwable t) {
logger.log(Level.WARNING, "Error closing connection with server: " + t.getMessage(), t);
}
@@ -192,4 +251,19 @@
return res;
}
+ private static final Map<ServerMetaData, ClusterMetaData> clusters = new ConcurrentHashMap<ServerMetaData, ClusterMetaData>();
+
+ private static void setClusterMetaData(ServerMetaData server, ClusterMetaData cluster) {
+ clusters.put(server, cluster);
+ }
+
+ private static ClusterMetaData getClusterMetaData(ServerMetaData server) {
+ ClusterMetaData cluster = clusters.get(server);
+ if (cluster == null) {
+ cluster = new ClusterMetaData(0, server.getLocation());
+ clusters.put(server, cluster);
+ }
+
+ return cluster;
+ }
}
Added: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterMetaData.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterMetaData.java?rev=693321&view=auto
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterMetaData.java (added)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterMetaData.java Mon Sep 8 17:09:25 2008
@@ -0,0 +1,84 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.client;
+
+import java.io.Externalizable;
+import java.io.ObjectInput;
+import java.io.IOException;
+import java.io.ObjectOutput;
+import java.net.URI;
+import java.net.URISyntaxException;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class ClusterMetaData implements Externalizable {
+ private URI[] locations;
+ private int version;
+ private String connectionStrategy;
+
+ public ClusterMetaData() {
+ }
+
+ public ClusterMetaData(int version, URI... locations) {
+ this.locations = locations;
+ this.version = version;
+ }
+
+ public URI[] getLocations() {
+ return locations;
+ }
+
+ public int getVersion() {
+ return version;
+ }
+
+ public String getConnectionStrategy() {
+ return connectionStrategy;
+ }
+
+ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ in.readByte(); // for future use to identify format of the data.
+
+ version = in.readInt();
+ connectionStrategy = (String) in.readObject();
+
+ int length = in.readInt();
+ locations = new URI[length];
+
+ for (int i = 0; i < locations.length; i++) {
+ Object o = in.readObject();
+ try {
+ locations[i] = new URI((String)o);
+ } catch (URISyntaxException e) {
+ throw (IOException) new IOException().initCause(e);
+ }
+ }
+ }
+
+ public void writeExternal(ObjectOutput out) throws IOException {
+ // write out the version of the serialized data for future use
+ out.writeByte(1);
+
+ out.writeInt(version);
+ out.writeObject(connectionStrategy);
+ out.writeInt(locations.length);
+ for (URI uri : locations) {
+ out.writeObject(uri.toString());
+ }
+ }
+}
Added: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterRequest.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterRequest.java?rev=693321&view=auto
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterRequest.java (added)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterRequest.java Mon Sep 8 17:09:25 2008
@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.client;
+
+import java.io.ObjectInput;
+import java.io.IOException;
+import java.io.ObjectOutput;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class ClusterRequest implements Request {
+ private int clusterMetaDataVersion;
+
+ public ClusterRequest() {
+ }
+
+ public ClusterRequest(ClusterMetaData clusterMetaData) {
+ clusterMetaDataVersion = clusterMetaData.getVersion();
+ }
+
+ public byte getRequestType() {
+ return RequestMethodConstants.CLUSTER_REQUEST;
+ }
+
+ public int getClusterMetaDataVersion() {
+ return clusterMetaDataVersion;
+ }
+
+ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ clusterMetaDataVersion = in.readInt();
+ }
+
+ public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeInt(clusterMetaDataVersion);
+ }
+}
Added: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterResponse.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterResponse.java?rev=693321&view=auto
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterResponse.java (added)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ClusterResponse.java Mon Sep 8 17:09:25 2008
@@ -0,0 +1,94 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.client;
+
+import java.io.ObjectInput;
+import java.io.IOException;
+import java.io.ObjectOutput;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class ClusterResponse implements Response {
+ public static enum Code {
+ CURRENT, UPDATE, FAILURE;
+ }
+
+ private Code responseCode;
+ private ClusterMetaData updatedMetaData;
+ private Throwable failure;
+
+ public ClusterResponse() {
+ }
+
+ public Code getResponseCode() {
+ return responseCode;
+ }
+
+ public void setCurrent() {
+ this.responseCode = Code.CURRENT;
+ }
+
+ public void setUpdatedMetaData(ClusterMetaData updatedMetaData) {
+ this.responseCode = Code.UPDATE;
+ this.updatedMetaData = updatedMetaData;
+ }
+
+ public ClusterMetaData getUpdatedMetaData() {
+ return updatedMetaData;
+ }
+
+ public Throwable getFailure() {
+ return failure;
+ }
+
+ public void setFailure(Throwable failure) {
+ this.responseCode = Code.FAILURE;
+ this.failure = failure;
+ }
+
+
+ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ byte i = in.readByte();
+ responseCode = Code.values()[i];
+
+ switch(responseCode){
+ case CURRENT: break;
+ case UPDATE: {
+ updatedMetaData = new ClusterMetaData();
+ updatedMetaData.readExternal(in);
+ }; break;
+ case FAILURE:{
+ failure = (IOException) in.readObject();
+ }
+ }
+ }
+
+ public void writeExternal(ObjectOutput out) throws IOException {
+ out.writeByte(responseCode.ordinal());
+
+ switch(responseCode){
+ case CURRENT: break;
+ case UPDATE: {
+ updatedMetaData.writeExternal(out);
+ }; break;
+ case FAILURE:{
+ out.writeObject(failure);
+ }
+ }
+ }
+}
Modified: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ConnectionManager.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ConnectionManager.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ConnectionManager.java (original)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ConnectionManager.java Mon Sep 8 17:09:25 2008
@@ -38,8 +38,8 @@
}
- public static Connection getConnection(ServerMetaData serverMetaData) throws IOException {
- String name = serverMetaData.getConnectionStrategy();
+ public static Connection getConnection(ClusterMetaData cluster, ServerMetaData server) throws IOException {
+ String name = cluster.getConnectionStrategy();
if (name == null) name = "default";
@@ -48,7 +48,7 @@
if (strategy == null) throw new IOException("Unsupported ConnectionStrategy \"" + name + "\"");
- return strategy.connect(serverMetaData);
+ return strategy.connect(cluster, server);
}
public static Connection getConnection(URI uri) throws IOException {
Modified: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ConnectionStrategy.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ConnectionStrategy.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ConnectionStrategy.java (original)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ConnectionStrategy.java Mon Sep 8 17:09:25 2008
@@ -19,5 +19,5 @@
import java.io.IOException;
public interface ConnectionStrategy {
- public Connection connect(ServerMetaData server) throws IOException;
+ public Connection connect(ClusterMetaData cluster, ServerMetaData server) throws IOException;
}
Modified: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/EJBRequest.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/EJBRequest.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/EJBRequest.java (original)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/EJBRequest.java Mon Sep 8 17:09:25 2008
@@ -482,6 +482,9 @@
case RequestMethodConstants.EJB_HOME_CREATE:
s = new StringBuffer("EJB_HOME.CREATE");
break;
+ case RequestMethodConstants.EJB_HOME_METHOD:
+ s = new StringBuffer("EJB_HOME.HOME_METHOD");
+ break;
case RequestMethodConstants.EJB_OBJECT_GET_EJB_HOME:
s = new StringBuffer("EJB_OBJECT.GET_EJB_HOME");
break;
@@ -500,6 +503,8 @@
case RequestMethodConstants.EJB_OBJECT_BUSINESS_METHOD:
s = new StringBuffer("EJB_OBJECT.BUSINESS_METHOD");
break;
+ default:
+ s = new StringBuffer("EJB_UKNOWN."+requestMethod);
}
s.append(':').append(deploymentId);
if (body != null) {
Added: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/KeepAliveStyle.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/KeepAliveStyle.java?rev=693321&view=auto
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/KeepAliveStyle.java (added)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/KeepAliveStyle.java Mon Sep 8 17:09:25 2008
@@ -0,0 +1,24 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.client;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public enum KeepAliveStyle {
+ PING, PING_PONG, PING_PING;
+}
Modified: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ProtocolMetaData.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ProtocolMetaData.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ProtocolMetaData.java (original)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ProtocolMetaData.java Mon Sep 8 17:09:25 2008
@@ -20,6 +20,7 @@
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.io.EOFException;
/**
* OpenEJB Enterprise Javabean Protocol (OEJP)
@@ -86,7 +87,7 @@
for (int i = 0; i < spec.length; i++) {
spec[i] = (byte) in.read();
if (spec[i] == -1){
- throw new IOException("Unable to read protocol version. Reached the end of the stream.");
+ throw new EOFException("Unable to read protocol version. Reached the end of the stream.");
}
}
init(new String(spec,"UTF-8"));
Modified: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/RequestMethodConstants.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/RequestMethodConstants.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/RequestMethodConstants.java (original)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/RequestMethodConstants.java Mon Sep 8 17:09:25 2008
@@ -23,6 +23,7 @@
public static final byte EJB_REQUEST = (byte) 0;
public static final byte JNDI_REQUEST = (byte) 1;
public static final byte AUTH_REQUEST = (byte) 2;
+ public static final byte CLUSTER_REQUEST = (byte) 3;
public static final byte STOP_REQUEST_Quit = (byte) 'Q';
public static final byte STOP_REQUEST_quit = (byte) 'q';
public static final byte STOP_REQUEST_Stop = (byte) 'S';
Modified: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ServerMetaData.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ServerMetaData.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ServerMetaData.java (original)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/ServerMetaData.java Mon Sep 8 17:09:25 2008
@@ -26,7 +26,6 @@
public class ServerMetaData implements Externalizable {
private transient URI[] locations;
- private transient String connectionStrategy;
public ServerMetaData() {
}
@@ -35,17 +34,12 @@
this.locations = locations;
}
- public ServerMetaData(String connectionStrategy, URI ... locations) {
- this.connectionStrategy = connectionStrategy;
- this.locations = locations;
- }
-
public void merge(ServerMetaData toMerge) {
locations = toMerge.locations;
}
- public String getConnectionStrategy() {
- return connectionStrategy;
+ public URI getLocation() {
+ return locations[0];
}
public URI[] getLocations() {
@@ -54,7 +48,7 @@
public int buildHash() {
int locationsHash = 0;
- for (URI location : locations) {
+ for (URI location : this.locations) {
locationsHash += location.hashCode();
}
return locationsHash;
@@ -63,18 +57,13 @@
public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
byte version = in.readByte();
- if (version > 1){
- connectionStrategy = (String) in.readObject();
- }
-
locations = (URI[]) in.readObject();
}
public void writeExternal(ObjectOutput out) throws IOException {
// write out the version of the serialized data for future use
- out.writeByte(2);
+ out.writeByte(1);
- out.writeObject(connectionStrategy);
out.writeObject(locations);
}
Modified: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/SocketConnectionFactory.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/SocketConnectionFactory.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/SocketConnectionFactory.java (original)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/SocketConnectionFactory.java Mon Sep 8 17:09:25 2008
@@ -25,7 +25,6 @@
import java.net.Socket;
import java.net.URI;
import java.net.ConnectException;
-import java.util.Properties;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -36,6 +35,8 @@
public class SocketConnectionFactory implements ConnectionFactory {
+ private KeepAliveStyle keepAliveStyle = KeepAliveStyle.PING;
+
private static Map<URI, SocketConnection> connections = new ConcurrentHashMap<URI, SocketConnection>();
public Connection getConnection(URI uri) throws java.io.IOException {
@@ -60,15 +61,48 @@
try {
conn.lock.tryLock(60*5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
+ Thread.interrupted();
throw new IOException("Connection busy");
}
OutputStream ouputStream = conn.getOuputStream();
- ouputStream.write(30);
-// ouputStream.flush();
+ if (conn.socket.isClosed()){
+ connections.remove(uri);
+ return getConnection(uri);
+ }
+
+ try {
+ ouputStream.write(getKeepAliveStyle().ordinal());
+
+ switch(getKeepAliveStyle()){
+ case PING_PING: {
+ ouputStream.flush();
+ ouputStream.write(getKeepAliveStyle().ordinal());
+ ouputStream.flush();
+ }
+ break;
+ case PING_PONG: {
+ ouputStream.flush();
+ conn.getInputStream().read();
+ }
+ }
+ } catch (IOException e) {
+ connections.remove(uri);
+ throw e;
+ }
+
return conn;
}
+ public KeepAliveStyle getKeepAliveStyle() {
+ String property = System.getProperty("openejb.client.keepalive");
+ if (property != null){
+ property = property.toUpperCase();
+ return KeepAliveStyle.valueOf(property);
+ }
+ return keepAliveStyle;
+ }
+
class SocketConnection implements Connection {
private Socket socket = null;
Modified: openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/StickyConnectionStrategy.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/StickyConnectionStrategy.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/StickyConnectionStrategy.java (original)
+++ openejb/trunk/openejb3/server/openejb-client/src/main/java/org/apache/openejb/client/StickyConnectionStrategy.java Mon Sep 8 17:09:25 2008
@@ -27,31 +27,38 @@
private URI lastLocation;
- public Connection connect(ServerMetaData server) throws IOException {
- URI[] locations = server.getLocations();
+ public Connection connect(ClusterMetaData cluster, ServerMetaData server) throws IOException {
+ URI[] locations = cluster.getLocations();
+ if (locations.length == 0){
+ return connect(server.getLocation());
+ }
if (null != lastLocation) {
- for (int i = 0; i < locations.length; i++) {
- if (locations[i].equals(lastLocation)) {
+ for (URI uri : locations) {
+ if (uri.equals(lastLocation)) {
try {
return connect(lastLocation);
} catch (IOException e) {
- LOGGER.log(Level.WARNING, "Cannot connect to last server: " + lastLocation.getHost() + ":" + lastLocation.getPort() + " Exception: ", e);
+ if (locations.length > 1){
+ LOGGER.log(Level.WARNING, "Failing over. Cannot connect to last server: " + lastLocation.toString() + " Exception: " + e.getClass().getName() +" " + e.getMessage());
+ }
}
}
}
}
Connection connection = null;
- for (int i = 0; i < locations.length; i++) {
- URI uri = locations[i];
+ for (URI uri : locations) {
+ if (uri.equals(lastLocation)) {
+ continue;
+ }
try {
connection = connect(uri);
lastLocation = uri;
break;
} catch (IOException e) {
- LOGGER.log(Level.WARNING, "Cannot connect to server(s): " + uri.getHost() + ":" + uri.getPort() + " Exception: ", e);
+ LOGGER.log(Level.WARNING, "Failover: Cannot connect to server(s): " + uri.toString() + " Exception: " + e.getMessage()+". Trying next.");
} catch (Throwable e) {
- throw new RemoteException("Cannot connect to server: " + uri.getHost() + ":" + uri.getPort() + " due to an unkown exception in the OpenEJB client: ", e);
+ throw new RemoteException("Failover: Cannot connect to server: " + uri.toString() + " due to an unkown exception in the OpenEJB client: ", e);
}
}
Modified: openejb/trunk/openejb3/server/openejb-client/src/test/java/org/apache/openejb/client/StickyConnectionStrategyTest.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-client/src/test/java/org/apache/openejb/client/StickyConnectionStrategyTest.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-client/src/test/java/org/apache/openejb/client/StickyConnectionStrategyTest.java (original)
+++ openejb/trunk/openejb3/server/openejb-client/src/test/java/org/apache/openejb/client/StickyConnectionStrategyTest.java Mon Sep 8 17:09:25 2008
@@ -18,7 +18,6 @@
import java.io.IOException;
import java.net.URI;
-import java.rmi.RemoteException;
import com.agical.rmock.extension.junit.RMockTestCase;
@@ -51,8 +50,8 @@
startVerification();
try {
- ServerMetaData server = new ServerMetaData(locations);
- factoryStrategy.connect(server);
+ ServerMetaData server = new ServerMetaData(locations[0]);
+ factoryStrategy.connect(new ClusterMetaData(1, locations), server);
fail();
} catch (IOException e) {
}
@@ -66,8 +65,9 @@
startVerification();
- ServerMetaData server = new ServerMetaData(locations);
- Connection actualConnection = factoryStrategy.connect(server);
+ ServerMetaData server = new ServerMetaData(locations[0]);
+ ClusterMetaData cluster = new ClusterMetaData(1, locations);
+ Connection actualConnection = factoryStrategy.connect(cluster, server);
assertSame(expectedConnection, actualConnection);
}
@@ -80,9 +80,10 @@
startVerification();
- ServerMetaData server = new ServerMetaData(locations);
- factoryStrategy.connect(server);
- factoryStrategy.connect(server);
+ ServerMetaData server = new ServerMetaData(locations[0]);
+ ClusterMetaData cluster = new ClusterMetaData(1, locations);
+ factoryStrategy.connect(cluster, server);
+ factoryStrategy.connect(cluster, server);
}
}
Modified: openejb/trunk/openejb3/server/openejb-discovery/src/main/java/org/apache/openejb/server/discovery/MulticastDiscoveryAgent.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-discovery/src/main/java/org/apache/openejb/server/discovery/MulticastDiscoveryAgent.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-discovery/src/main/java/org/apache/openejb/server/discovery/MulticastDiscoveryAgent.java (original)
+++ openejb/trunk/openejb3/server/openejb-discovery/src/main/java/org/apache/openejb/server/discovery/MulticastDiscoveryAgent.java Mon Sep 8 17:09:25 2008
@@ -22,7 +22,8 @@
import org.apache.openejb.server.SelfManaging;
import org.apache.openejb.server.ServerService;
import org.apache.openejb.server.ServiceException;
-import org.apache.openejb.server.ServiceDaemon;
+import org.apache.openejb.server.DiscoveryAgent;
+import org.apache.openejb.server.DiscoveryListener;
import org.apache.openejb.util.LogCategory;
import org.apache.openejb.util.Logger;
@@ -72,6 +73,9 @@
private Map<String, Service> registeredServices = new ConcurrentHashMap<String, Service>();
+ private String group = "default";
+ private String groupPrefix = group + ":";
+
private int maxMissedHeartbeats = 10;
private long heartRate = 500;
@@ -83,9 +87,9 @@
// ---------------------------------
// Listenting specific settings
- private long initialReconnectDelay = 1000 * 5;
+ private long reconnectDelay = 1000 * 5;
private long maxReconnectDelay = 1000 * 30;
- private long backOffMultiplier = 0;
+ private long exponentialBackoff = 0;
private boolean useExponentialBackOff;
private int maxReconnectAttempts = 10; // todo: check this out
// ---------------------------------
@@ -94,19 +98,21 @@
public void init(Properties props) throws Exception {
host = props.getProperty("bind", host);
+ group = props.getProperty("group", group);
+ groupPrefix = group + ":";
port = getInt(props, "port", port);
heartRate = getLong(props, "heart_rate", heartRate);
maxMissedHeartbeats = getInt(props, "max_missed_heartbeats", maxMissedHeartbeats);
- loopbackMode = getBoolean(props, "max_missed_heartbeats", loopbackMode);
+ loopbackMode = getBoolean(props, "loopback_mode", loopbackMode);
- initialReconnectDelay = getLong(props, "reconnect_delay", initialReconnectDelay);
- maxReconnectDelay = getLong(props, "max_reconnect_delay", initialReconnectDelay);
+ reconnectDelay = getLong(props, "reconnect_delay", reconnectDelay);
+ maxReconnectDelay = getLong(props, "max_reconnect_delay", reconnectDelay);
maxReconnectAttempts = getInt(props, "max_reconnect_attempts", maxReconnectAttempts);
- backOffMultiplier = getLong(props, "exponential_backoff", backOffMultiplier);
+ exponentialBackoff = getLong(props, "exponential_backoff", exponentialBackoff);
- useExponentialBackOff = (backOffMultiplier > 1);
+ useExponentialBackOff = (exponentialBackoff > 1);
}
public String getIP() {
@@ -127,12 +133,14 @@
public void registerService(URI serviceUri) throws IOException {
Service service = new Service(serviceUri);
- this.registeredServices.put(service.uriString, service);
+ this.registeredServices.put(service.broadcastString, service);
+ this.listener.fireServiceAddedEvent(serviceUri);
}
public void unregisterService(URI serviceUri) throws IOException {
Service service = new Service(serviceUri);
- this.registeredServices.remove(service.uriString);
+ this.registeredServices.remove(service.broadcastString);
+ this.listener.fireServiceRemovedEvent(serviceUri);
}
public void reportFailed(URI serviceUri) throws IOException {
@@ -141,7 +149,7 @@
private boolean isSelf(Service service) {
- return isSelf(service.uriString);
+ return isSelf(service.broadcastString);
}
private boolean isSelf(String service) {
@@ -204,15 +212,18 @@
class Service {
private final URI uri;
- private final String uriString;
+ private final String broadcastString;
public Service(URI uri) {
this.uri = uri;
- this.uriString = uri.toString();
+ this.broadcastString = groupPrefix + uri.toString();
}
public Service(String uriString) throws URISyntaxException {
- this(new URI(uriString));
+ URI uri = new URI(uriString);
+ uri = new URI(uri.getSchemeSpecificPart());
+ this.uri = uri;
+ this.broadcastString = uriString;
}
}
@@ -253,23 +264,23 @@
dead = true;
failureCount++;
- long reconnectDelay;
+ long delay;
if (useExponentialBackOff) {
- reconnectDelay = (long) Math.pow(backOffMultiplier, failureCount);
- if (reconnectDelay > maxReconnectDelay) {
- reconnectDelay = maxReconnectDelay;
+ delay = (long) Math.pow(exponentialBackoff, failureCount);
+ if (delay > maxReconnectDelay) {
+ delay = maxReconnectDelay;
}
} else {
- reconnectDelay = initialReconnectDelay;
+ delay = reconnectDelay;
}
if (log.isDebugEnabled()) {
log.debug("Remote failure of " + service + " while still receiving multicast advertisements. " +
- "Advertising events will be suppressed for " + reconnectDelay
+ "Advertising events will be suppressed for " + delay
+ " ms, the current failure count is: " + failureCount);
}
- recoveryTime = System.currentTimeMillis() + reconnectDelay;
+ recoveryTime = System.currentTimeMillis() + delay;
return true;
}
return false;
@@ -344,6 +355,11 @@
if (discoveryListener == null) {
return;
}
+
+ if (!uriString.startsWith(groupPrefix)){
+ return;
+ }
+
if (isSelf(uriString)) {
return;
}
@@ -356,7 +372,7 @@
discoveredServices.put(uriString, vitals);
- fireServiceAddEvent(vitals.service.uri);
+ fireServiceAddedEvent(vitals.service.uri);
} catch (URISyntaxException e) {
// don't continuously log this
}
@@ -365,7 +381,7 @@
vitals.heartbeat();
if (vitals.doRecovery()) {
- fireServiceAddEvent(vitals.service.uri);
+ fireServiceAddedEvent(vitals.service.uri);
}
}
}
@@ -375,7 +391,7 @@
for (ServiceVitals serviceVitals : discoveredServices.values()) {
if (serviceVitals.getLastHeartbeat() < expireTime && !isSelf(serviceVitals.service)) {
- ServiceVitals vitals = discoveredServices.remove(serviceVitals.service.uriString);
+ ServiceVitals vitals = discoveredServices.remove(serviceVitals.service.broadcastString);
if (vitals != null && !vitals.isDead()) {
fireServiceRemovedEvent(vitals.service.uri);
}
@@ -408,7 +424,7 @@
}
}
- private void fireServiceAddEvent(final URI uri) {
+ private void fireServiceAddedEvent(final URI uri) {
if (discoveryListener != null) {
final DiscoveryListener discoveryListener = this.discoveryListener;
@@ -427,7 +443,7 @@
public void reportFailed(URI serviceUri) {
final Service service = new Service(serviceUri);
- ServiceVitals serviceVitals = discoveredServices.get(service.uriString);
+ ServiceVitals serviceVitals = discoveredServices.get(service.broadcastString);
if (serviceVitals != null && serviceVitals.pronounceDead()) {
fireServiceRemovedEvent(service.uri);
}
@@ -468,4 +484,91 @@
}
}
+
+ //
+ // Ordinary getters/setters
+ //
+
+ public long getExponentialBackoff() {
+ return exponentialBackoff;
+ }
+
+ public void setExponentialBackoff(long exponentialBackoff) {
+ this.exponentialBackoff = exponentialBackoff;
+ this.useExponentialBackOff = (exponentialBackoff > 1);
+ }
+
+ public String getGroup() {
+ return group;
+ }
+
+ public void setGroup(String group) {
+ this.group = group;
+ groupPrefix = group + ":";
+ }
+
+ public long getHeartRate() {
+ return heartRate;
+ }
+
+ public void setHeartRate(long heartRate) {
+ this.heartRate = heartRate;
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public void setHost(String host) {
+ this.host = host;
+ }
+
+ public long getReconnectDelay() {
+ return reconnectDelay;
+ }
+
+ public void setReconnectDelay(long reconnectDelay) {
+ this.reconnectDelay = reconnectDelay;
+ }
+
+ public boolean isLoopbackMode() {
+ return loopbackMode;
+ }
+
+ public void setLoopbackMode(boolean loopbackMode) {
+ this.loopbackMode = loopbackMode;
+ }
+
+ public int getMaxMissedHeartbeats() {
+ return maxMissedHeartbeats;
+ }
+
+ public void setMaxMissedHeartbeats(int maxMissedHeartbeats) {
+ this.maxMissedHeartbeats = maxMissedHeartbeats;
+ }
+
+ public int getMaxReconnectAttempts() {
+ return maxReconnectAttempts;
+ }
+
+ public void setMaxReconnectAttempts(int maxReconnectAttempts) {
+ this.maxReconnectAttempts = maxReconnectAttempts;
+ }
+
+ public long getMaxReconnectDelay() {
+ return maxReconnectDelay;
+ }
+
+ public void setMaxReconnectDelay(long maxReconnectDelay) {
+ this.maxReconnectDelay = maxReconnectDelay;
+ }
+
+ public int getTimeToLive() {
+ return timeToLive;
+ }
+
+ public void setTimeToLive(int timeToLive) {
+ this.timeToLive = timeToLive;
+ }
+
}
Modified: openejb/trunk/openejb3/server/openejb-discovery/src/test/java/org/apache/openejb/server/discovery/MulticastDiscoveryAgentTest.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-discovery/src/test/java/org/apache/openejb/server/discovery/MulticastDiscoveryAgentTest.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-discovery/src/test/java/org/apache/openejb/server/discovery/MulticastDiscoveryAgentTest.java (original)
+++ openejb/trunk/openejb3/server/openejb-discovery/src/test/java/org/apache/openejb/server/discovery/MulticastDiscoveryAgentTest.java Mon Sep 8 17:09:25 2008
@@ -25,6 +25,7 @@
import java.util.Set;
import org.apache.openejb.server.ServiceException;
+import org.apache.openejb.server.DiscoveryListener;
/**
* @version $Rev$ $Date$
Modified: openejb/trunk/openejb3/server/openejb-ejbd/pom.xml
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-ejbd/pom.xml?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-ejbd/pom.xml (original)
+++ openejb/trunk/openejb3/server/openejb-ejbd/pom.xml Mon Sep 8 17:09:25 2008
@@ -61,7 +61,7 @@
<forkMode>pertest</forkMode>
<!-- DEBUG: Uncomment this line and comment out the next -->
<!--<argLine>-Xdebug -Xnoagent -Djava.compiler=NONE -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5005 -javaagent:${project.build.directory}/openejb-javaagent-${version}.jar=foo=bar</argLine>-->
- <argLine>"-javaagent:${project.build.directory}/openejb-javaagent-${version}.jar=foo=bar"</argLine>
+ <argLine>"-javaagent:${project.build.directory}/openejb-javaagent-${version}.jar=foo=bar"</argLine>
<!--
<systemProperties>
<property>
@@ -69,7 +69,7 @@
<value>${project.build.directory}/test-classes</value>
</property>
</systemProperties>
- <basedir>${project.build.directory}</basedir>
+ <basedir>${project.build.directory}</basedir>
-->
</configuration>
</plugin>
@@ -103,6 +103,11 @@
</dependency>
<dependency>
<groupId>org.apache.openejb</groupId>
+ <artifactId>openejb-discovery</artifactId>
+ <version>${version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.openejb</groupId>
<artifactId>openejb-client</artifactId>
<version>${version}</version>
</dependency>
@@ -112,12 +117,12 @@
<version>${version}</version>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.apache.openejb</groupId>
- <artifactId>openejb-itests-beans</artifactId>
- <version>${version}</version>
- <scope>test</scope>
- </dependency>
+ <dependency>
+ <groupId>org.apache.openejb</groupId>
+ <artifactId>openejb-itests-beans</artifactId>
+ <version>${version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
Modified: openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/CallContext.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/CallContext.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/CallContext.java (original)
+++ openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/CallContext.java Mon Sep 8 17:09:25 2008
@@ -18,6 +18,7 @@
import org.apache.openejb.DeploymentInfo;
import org.apache.openejb.client.EJBRequest;
+import org.apache.openejb.client.ServerMetaData;
import java.util.HashMap;
@@ -49,6 +50,15 @@
set(DeploymentInfo.class, info);
}
+
+ public void setServerMetaData(ServerMetaData serverMetaData){
+ set(ServerMetaData.class, serverMetaData);
+ }
+
+ public ServerMetaData getServerMetaData(){
+ return get(ServerMetaData.class);
+ }
+
public EJBRequest getEJBRequest() {
return get(EJBRequest.class);
}
Modified: openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/ClientObjectFactory.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/ClientObjectFactory.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/ClientObjectFactory.java (original)
+++ openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/ClientObjectFactory.java Mon Sep 8 17:09:25 2008
@@ -30,13 +30,16 @@
import org.apache.openejb.client.ServerMetaData;
class ClientObjectFactory implements org.apache.openejb.spi.ApplicationServer {
- protected ServerMetaData sMetaData;
+
+ public static final ThreadLocal<ServerMetaData> serverMetaData = new ThreadLocal<ServerMetaData>();
+
+ protected ServerMetaData defaultServerMetaData;
public ClientObjectFactory(EjbDaemon daemon, Properties props) {
try {
String uriString = props.getProperty("openejb.ejbd.uri", "foo://127.0.0.1:4201");
- this.sMetaData = new ServerMetaData(new URI(uriString));
+ this.defaultServerMetaData = new ServerMetaData(new URI(uriString));
} catch (Exception e) {
e.printStackTrace();
}
@@ -78,11 +81,19 @@
idCode, null);
Object primKey = info.getPrimaryKey();
- EJBObjectHandler hanlder = EJBObjectHandler.createEJBObjectHandler(eMetaData, sMetaData, cMetaData, primKey);
+ EJBObjectHandler hanlder = EJBObjectHandler.createEJBObjectHandler(eMetaData, getServerMetaData(), cMetaData, primKey);
return new EJBObjectHandle(hanlder.createEJBObjectProxy());
}
+ private ServerMetaData getServerMetaData() {
+ ServerMetaData serverMetaData = ClientObjectFactory.serverMetaData.get();
+ if (serverMetaData == null){
+ serverMetaData = defaultServerMetaData;
+ }
+ return serverMetaData;
+ }
+
public javax.ejb.HomeHandle getHomeHandle(ProxyInfo info) {
CallContext call = CallContext.getCallContext();
DeploymentInfo deployment = info.getDeploymentInfo();
@@ -103,7 +114,7 @@
deployment.getDeploymentID().toString(),
idCode, null);
- EJBHomeHandler hanlder = EJBHomeHandler.createEJBHomeHandler(eMetaData, sMetaData, cMetaData);
+ EJBHomeHandler hanlder = EJBHomeHandler.createEJBHomeHandler(eMetaData, getServerMetaData(), cMetaData);
return new EJBHomeHandle(hanlder.createEJBHomeProxy());
}
@@ -129,7 +140,7 @@
idCode, null);
Object primKey = info.getPrimaryKey();
- EJBObjectHandler hanlder = EJBObjectHandler.createEJBObjectHandler(eMetaData, sMetaData, cMetaData, primKey);
+ EJBObjectHandler hanlder = EJBObjectHandler.createEJBObjectHandler(eMetaData, getServerMetaData(), cMetaData, primKey);
return (javax.ejb.EJBObject) hanlder.createEJBObjectProxy();
}
@@ -154,7 +165,7 @@
idCode, info.getInterfaces());
Object primKey = info.getPrimaryKey();
- EJBObjectHandler hanlder = EJBObjectHandler.createEJBObjectHandler(eMetaData, sMetaData, cMetaData, primKey);
+ EJBObjectHandler hanlder = EJBObjectHandler.createEJBObjectHandler(eMetaData, getServerMetaData(), cMetaData, primKey);
return hanlder.createEJBObjectProxy();
}
@@ -179,7 +190,7 @@
deployment.getDeploymentID().toString(),
idCode, null);
- EJBHomeHandler hanlder = EJBHomeHandler.createEJBHomeHandler(eMetaData, sMetaData, cMetaData);
+ EJBHomeHandler hanlder = EJBHomeHandler.createEJBHomeHandler(eMetaData, getServerMetaData(), cMetaData);
return hanlder.createEJBHomeProxy();
}
Added: openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/ClusterRequestHandler.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/ClusterRequestHandler.java?rev=693321&view=auto
==============================================================================
--- openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/ClusterRequestHandler.java (added)
+++ openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/ClusterRequestHandler.java Mon Sep 8 17:09:25 2008
@@ -0,0 +1,186 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.server.ejbd;
+
+import org.apache.openejb.client.ClusterRequest;
+import org.apache.openejb.client.ClusterResponse;
+import org.apache.openejb.client.ClusterMetaData;
+import org.apache.openejb.server.DiscoveryListener;
+import org.apache.openejb.util.LogCategory;
+import org.apache.openejb.util.Logger;
+import org.apache.openejb.util.Exceptions;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.net.URISyntaxException;
+import java.net.URI;
+import java.util.LinkedHashSet;
+import java.util.Set;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class ClusterRequestHandler implements DiscoveryListener {
+
+ private static final Logger logger = Logger.getInstance(LogCategory.OPENEJB_SERVER_REMOTE.createChild("cluster"), ClusterRequestHandler.class);
+
+ private final Data data = new Data();
+
+ public ClusterRequestHandler(EjbDaemon daemon) {
+ }
+
+
+ public void processRequest(ObjectInputStream in, ObjectOutputStream out) throws IOException {
+ ClusterRequest req = new ClusterRequest();
+ ClusterResponse res = new ClusterResponse();
+
+ try {
+ req.readExternal(in);
+ } catch (IOException e) {
+ res.setFailure(e);
+ sendErrorResponse("Cannot read ClusterRequest", e, res, out);
+ throw e;
+ } catch (ClassNotFoundException e) {
+ res.setFailure(e);
+ sendErrorResponse("Cannot read ClusterRequest", e, res, out);
+ throw (IOException) new IOException().initCause(e);
+ }
+
+ ClusterMetaData currentClusterMetaData = data.current();
+
+ if (req.getClusterMetaDataVersion() < currentClusterMetaData.getVersion()){
+ res.setUpdatedMetaData(currentClusterMetaData);
+ } else {
+ res.setCurrent();
+ }
+
+ try {
+ res.writeExternal(out);
+ } catch (IOException e) {
+ logger.error("Failed to write to ClusterResponse", e);
+ throw e;
+ }
+ }
+
+ private void sendErrorResponse(String message, Throwable t, ClusterResponse res, ObjectOutputStream out) throws IOException {
+ logger.fatal(message, t);
+ t = new IOException("The server has encountered a fatal error: " + message + " " + t).initCause(t);
+ try {
+ res.writeExternal(out);
+ } catch (IOException ie) {
+ String m = "Failed to write to ClusterResponse";
+ logger.error(m, ie);
+ throw Exceptions.newIOException(m, ie);
+ }
+ }
+
+ public void serviceAdded(final URI uri) {
+ try {
+ URI type = uri;
+ URI service = unwrap(type);
+
+ if ("ejb".equals(type.getScheme())) {
+ data.add(service);
+ }
+ } catch (URISyntaxException e) {
+ logger.error("serviceAdded: Invalid service URI format. Expected <group>:<type>:<serverURI> but found '" + uri.toString() + "'");
+ }
+ }
+
+ private URI unwrap(URI uri) throws URISyntaxException {
+ return new URI(uri.getSchemeSpecificPart());
+ }
+
+
+ public void serviceRemoved(final URI uri) {
+ try {
+ URI type = uri;
+ URI service = unwrap(type);
+
+ if ("ejb".equals(type.getScheme())) {
+ data.remove(service);
+ }
+ } catch (URISyntaxException e) {
+ logger.error("serviceAdded: Invalid service URI format. Expected <group>:<type>:<serverURI> but found '" + uri.toString() + "'");
+ }
+ }
+
+ private static class Data {
+ private ClusterMetaData current;
+ private ReadWriteLock sync = new ReentrantReadWriteLock();
+ private final java.util.Set set = new LinkedHashSet();
+
+ public Data() {
+ this.current = new ClusterMetaData(0);
+ }
+
+ public boolean add(URI o) {
+ Lock lock = sync.writeLock();
+ lock.lock();
+ ClusterMetaData nextVersion = null;
+ try {
+ if (set.add(o)) {
+ nextVersion = newClusterMetaData(set, current);
+ return true;
+ } else {
+ return false;
+ }
+ } finally {
+ if (nextVersion != null) current = nextVersion;
+ lock.unlock();
+ }
+ }
+
+ public boolean remove(Object o) {
+ Lock lock = sync.writeLock();
+ lock.lock();
+ ClusterMetaData nextVersion = null;
+ try {
+ if (set.remove(o)) {
+ nextVersion = newClusterMetaData(set, current);
+ return true;
+ } else {
+ return false;
+ }
+ } finally {
+ if (nextVersion != null) current = nextVersion;
+ lock.unlock();
+ }
+ }
+
+ private static ClusterMetaData newClusterMetaData(Set set, ClusterMetaData current) {
+ URI[] locations = new URI[set.size()];
+ set.toArray(locations);
+ return new ClusterMetaData(current.getVersion()+1, locations);
+ }
+
+ public ClusterMetaData current() {
+ Lock lock = sync.readLock();
+ lock.lock();
+ try {
+ return current;
+ } finally {
+ lock.unlock();
+ }
+ }
+ }
+
+}
Modified: openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbDaemon.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbDaemon.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbDaemon.java (original)
+++ openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbDaemon.java Mon Sep 8 17:09:25 2008
@@ -27,19 +27,21 @@
import org.apache.openejb.DeploymentInfo;
import org.apache.openejb.ProxyInfo;
+import org.apache.openejb.server.DiscoveryAgent;
import org.apache.openejb.loader.SystemInstance;
import org.apache.openejb.spi.ContainerSystem;
import org.apache.openejb.client.EJBRequest;
import org.apache.openejb.client.RequestMethodConstants;
import org.apache.openejb.client.EjbObjectInputStream;
import org.apache.openejb.client.ProtocolMetaData;
+import org.apache.openejb.client.ServerMetaData;
import org.apache.openejb.util.LogCategory;
import org.apache.openejb.util.Logger;
import org.apache.openejb.util.Messages;
public class EjbDaemon implements org.apache.openejb.spi.ApplicationServer {
- private static final ProtocolMetaData PROTOCOL_VERSION = new ProtocolMetaData("3.0");
+ private static final ProtocolMetaData PROTOCOL_VERSION = new ProtocolMetaData("3.1");
private static final Messages _messages = new Messages("org.apache.openejb.server.util.resources");
static final Logger logger = Logger.getInstance(LogCategory.OPENEJB_SERVER_REMOTE, "org.apache.openejb.server.util.resources");
@@ -49,6 +51,7 @@
private EjbRequestHandler ejbHandler;
private JndiRequestHandler jndiHandler;
private AuthRequestHandler authHandler;
+ private ClusterRequestHandler clusterHandler;
boolean stop = false;
@@ -74,6 +77,12 @@
ejbHandler = new EjbRequestHandler(this);
jndiHandler = new JndiRequestHandler(this);
authHandler = new AuthRequestHandler(this);
+ clusterHandler = new ClusterRequestHandler(this);
+
+ DiscoveryAgent discovery = SystemInstance.get().getComponent(DiscoveryAgent.class);
+ if (discovery != null) {
+ discovery.setDiscoveryListener(clusterHandler);
+ }
}
public void service(Socket socket) throws IOException {
@@ -100,18 +109,36 @@
try {
+ // Read Protocol Version
protocolMetaData.readExternal(in);
PROTOCOL_VERSION.writeExternal(out);
- byte requestType = (byte) in.read();
+ ois = new EjbObjectInputStream(in);
+ oos = new ObjectOutputStream(out);
+
+ // Read ServerMetaData
+ ServerMetaData serverMetaData = new ServerMetaData();
+ serverMetaData.readExternal(ois);
+ ClientObjectFactory.serverMetaData.set(serverMetaData);
+
+ // Read request type
+ byte requestType = (byte) ois.read();
+
+ if (requestType == -1) {
+ return;
+ }
+
+ if (requestType == RequestMethodConstants.CLUSTER_REQUEST){
+ processClusterRequest(ois, oos);
+ }
+
+ requestType = (byte) ois.read();
if (requestType == -1) {
return;
}
- ois = new EjbObjectInputStream(in);
- oos = new ObjectOutputStream(out);
// Exceptions should not be thrown from these methods
// They should handle their own exceptions and clean
@@ -139,6 +166,7 @@
} catch (Throwable e) {
logger.error("\""+requestTypeName +" "+ protocolMetaData.getSpec() + "\" FAIL \"Unexpected error - "+e.getMessage()+"\"",e);
} finally {
+ ClientObjectFactory.serverMetaData.remove();
try {
if (oos != null) {
oos.flush();
@@ -153,6 +181,10 @@
}
}
+ private void processClusterRequest(ObjectInputStream in, ObjectOutputStream out) throws IOException {
+ clusterHandler.processRequest(in, out);
+ }
+
protected DeploymentInfo getDeployment(EJBRequest req) throws RemoteException {
String deploymentId = req.getDeploymentId();
DeploymentInfo deploymentInfo = containerSystem.getDeploymentInfo(deploymentId);
Modified: openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbServer.java
URL: http://svn.apache.org/viewvc/openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbServer.java?rev=693321&r1=693320&r2=693321&view=diff
==============================================================================
--- openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbServer.java (original)
+++ openejb/trunk/openejb3/server/openejb-ejbd/src/main/java/org/apache/openejb/server/ejbd/EjbServer.java Mon Sep 8 17:09:25 2008
@@ -47,9 +47,11 @@
}
public void start() throws ServiceException {
+ keepAlive.start();
}
public void stop() throws ServiceException {
+ keepAlive.stop();
}
public String getName() {