You are viewing a plain text version of this content. The canonical link for it is here.
Posted to scm@geronimo.apache.org by jg...@apache.org on 2006/09/17 14:35:47 UTC
svn commit: r447058 [3/4] - in /geronimo/sandbox/gcache/openwire: ./
src/main/java/org/apache/geronimo/openwire/command/
src/main/java/org/apache/geronimo/openwire/state/
src/main/java/org/apache/geronimo/openwire/thread/
src/main/java/org/apache/geron...
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/thread/Valve.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/thread/Valve.java?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/thread/Valve.java (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/thread/Valve.java Sun Sep 17 05:35:45 2006
@@ -0,0 +1,107 @@
+/**
+ *
+ * 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.geronimo.openwire.thread;
+
+/**
+ * A Valve is a synchronization object used enable or disable the "flow" of concurrent
+ * processing.
+ *
+ *
+ * @version $Revision$
+ */
+final public class Valve {
+
+ private final Object mutex = new Object();
+ private boolean on;
+ private int turningOff=0;
+ private int usage=0;
+
+ public Valve(boolean on) {
+ this.on = on;
+ }
+
+ /**
+ * Turns the valve on. This method blocks until the valve is off.
+ * @throws InterruptedException
+ */
+ public void turnOn() throws InterruptedException {
+ synchronized(mutex) {
+ while( on ) {
+ mutex.wait();
+ }
+ on=true;
+ mutex.notifyAll();
+ }
+ }
+
+ boolean isOn() {
+ synchronized(mutex) {
+ return on;
+ }
+ }
+
+ /**
+ * Turns the valve off. This method blocks until the valve is on and the valve is not
+ * in use.
+ *
+ * @throws InterruptedException
+ */
+ public void turnOff() throws InterruptedException {
+ synchronized(mutex) {
+ try {
+ ++turningOff;
+ while( usage > 0 || !on) {
+ mutex.wait();
+ }
+ on=false;
+ mutex.notifyAll();
+ } finally {
+ --turningOff;
+ }
+ }
+ }
+
+ /**
+ * Increments the use counter of the valve. This method blocks if the valve is off,
+ * or is being turned off.
+ *
+ * @throws InterruptedException
+ */
+ public void increment() throws InterruptedException {
+ synchronized(mutex) {
+ // Do we have to wait for the value to be on?
+ while( turningOff>0 || !on ) {
+ mutex.wait();
+ }
+ usage++;
+ }
+ }
+
+ /**
+ * Decrements the use counter of the valve.
+ */
+ public void decrement() {
+ synchronized(mutex) {
+ usage--;
+ if( turningOff>0 && usage < 1 ) {
+ mutex.notifyAll();
+ }
+ }
+ }
+
+}
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/thread/Valve.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/thread/Valve.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/thread/Valve.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/InactivityMonitor.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/InactivityMonitor.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/InactivityMonitor.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/InactivityMonitor.java Sun Sep 17 05:35:45 2006
@@ -21,9 +21,9 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.geronimo.openwire.command.WireFormatInfo;
-import org.apache.geronimo.openwire.command.KeepAliveInfo;
import org.apache.geronimo.openwire.command.Command;
+import org.apache.geronimo.openwire.command.KeepAliveInfo;
+import org.apache.geronimo.openwire.command.WireFormatInfo;
import org.apache.geronimo.openwire.thread.Scheduler;
import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MarshallingTransportFilter.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MarshallingTransportFilter.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MarshallingTransportFilter.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MarshallingTransportFilter.java Sun Sep 17 05:35:45 2006
@@ -17,10 +17,10 @@
*/
package org.apache.geronimo.openwire.transport;
-import org.apache.geronimo.openwire.wireformat.WireFormat;
-import org.apache.geronimo.openwire.command.Command;
-
import java.io.IOException;
+
+import org.apache.geronimo.openwire.command.Command;
+import org.apache.geronimo.openwire.wireformat.WireFormat;
public class MarshallingTransportFilter extends TransportFilter {
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MutexTransport.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MutexTransport.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MutexTransport.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MutexTransport.java Sun Sep 17 05:35:45 2006
@@ -17,10 +17,10 @@
*/
package org.apache.geronimo.openwire.transport;
-import org.apache.geronimo.openwire.command.Response;
-import org.apache.geronimo.openwire.command.Command;
-
import java.io.IOException;
+
+import org.apache.geronimo.openwire.command.Command;
+import org.apache.geronimo.openwire.command.Response;
/**
* @version $Revision$
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MutexTransport.java.orig
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MutexTransport.java.orig?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MutexTransport.java.orig (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MutexTransport.java.orig Sun Sep 17 05:35:45 2006
@@ -0,0 +1,64 @@
+/**
+ *
+ * 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.geronimo.openwire.transport;
+
+import org.apache.geronimo.openwire.command.Response;
+import org.apache.geronimo.openwire.command.Command;
+
+import java.io.IOException;
+
+/**
+ * @version $Revision: 446644 $
+ */
+public class MutexTransport extends TransportFilter {
+
+ private final Object writeMutex = new Object();
+
+ public MutexTransport(Transport next) {
+ super(next);
+ }
+
+ public FutureResponse asyncRequest(Command command, ResponseCallback responseCallback) throws IOException {
+ synchronized(writeMutex) {
+ return next.asyncRequest(command, null);
+ }
+ }
+
+ public void oneway(Command command) throws IOException {
+ synchronized(writeMutex) {
+ next.oneway(command);
+ }
+ }
+
+ public Response request(Command command) throws IOException {
+ synchronized(writeMutex) {
+ return next.request(command);
+ }
+ }
+
+ public Response request(Command command,int timeout) throws IOException {
+ synchronized(writeMutex){
+ return next.request(command,timeout);
+ }
+ }
+
+ public String toString() {
+ return next.toString();
+ }
+
+}
\ No newline at end of file
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/MutexTransport.java.orig
------------------------------------------------------------------------------
svn:executable = *
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/ResponseCorrelator.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/ResponseCorrelator.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/ResponseCorrelator.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/ResponseCorrelator.java Sun Sep 17 05:35:45 2006
@@ -24,8 +24,8 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.geronimo.openwire.command.Command;
-import org.apache.geronimo.openwire.command.Response;
import org.apache.geronimo.openwire.command.ExceptionResponse;
+import org.apache.geronimo.openwire.command.Response;
import org.apache.geronimo.openwire.util.IntSequenceGenerator;
import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/Transport.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/Transport.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/Transport.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/Transport.java Sun Sep 17 05:35:45 2006
@@ -17,11 +17,11 @@
*/
package org.apache.geronimo.openwire.transport;
+import java.io.IOException;
+
+import org.apache.geronimo.openwire.Service;
import org.apache.geronimo.openwire.command.Command;
import org.apache.geronimo.openwire.command.Response;
-import org.apache.geronimo.openwire.Service;
-
-import java.io.IOException;
/**
* Represents the client side of a transport allowing messages
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportFactory.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportFactory.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportFactory.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportFactory.java Sun Sep 17 05:35:45 2006
@@ -26,9 +26,9 @@
import java.util.Map;
import org.apache.geronimo.openwire.util.FactoryFinder;
-import org.apache.geronimo.openwire.util.URISupport;
import org.apache.geronimo.openwire.util.IOExceptionSupport;
import org.apache.geronimo.openwire.util.IntrospectionSupport;
+import org.apache.geronimo.openwire.util.URISupport;
import org.apache.geronimo.openwire.wireformat.WireFormat;
import org.apache.geronimo.openwire.wireformat.WireFormatFactory;
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportFilter.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportFilter.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportFilter.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportFilter.java Sun Sep 17 05:35:45 2006
@@ -18,6 +18,7 @@
package org.apache.geronimo.openwire.transport;
import java.io.IOException;
+
import org.apache.geronimo.openwire.command.Command;
import org.apache.geronimo.openwire.command.Response;
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportLogger.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportLogger.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportLogger.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportLogger.java Sun Sep 17 05:35:45 2006
@@ -21,8 +21,8 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.geronimo.openwire.command.Response;
import org.apache.geronimo.openwire.command.Command;
+import org.apache.geronimo.openwire.command.Response;
/**
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportServerSupport.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportServerSupport.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportServerSupport.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportServerSupport.java Sun Sep 17 05:35:45 2006
@@ -17,9 +17,9 @@
*/
package org.apache.geronimo.openwire.transport;
-import org.apache.geronimo.openwire.util.ServiceSupport;
-
import java.net.URI;
+
+import org.apache.geronimo.openwire.util.ServiceSupport;
/**
* A useful base class for implementations of {@link TransportServer}
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportServerThreadSupport.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportServerThreadSupport.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportServerThreadSupport.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportServerThreadSupport.java Sun Sep 17 05:35:45 2006
@@ -18,12 +18,12 @@
package org.apache.geronimo.openwire.transport;
+import java.net.URI;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.geronimo.openwire.util.ServiceStopper;
import org.apache.geronimo.openwire.ThreadPriorities;
-
-import java.net.URI;
+import org.apache.geronimo.openwire.util.ServiceStopper;
/**
* A useful base class for implementations of {@link TransportServer} which uses
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportSupport.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportSupport.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportSupport.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/TransportSupport.java Sun Sep 17 05:35:45 2006
@@ -17,13 +17,13 @@
*/
package org.apache.geronimo.openwire.transport;
+import java.io.IOException;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.geronimo.openwire.util.ServiceSupport;
import org.apache.geronimo.openwire.command.Command;
import org.apache.geronimo.openwire.command.Response;
-
-import java.io.IOException;
+import org.apache.geronimo.openwire.util.ServiceSupport;
/**
* A useful base class for transport implementations.
Modified: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/WireFormatNegotiator.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/WireFormatNegotiator.java?view=diff&rev=447058&r1=447057&r2=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/WireFormatNegotiator.java (original)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/WireFormatNegotiator.java Sun Sep 17 05:35:45 2006
@@ -22,8 +22,8 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.geronimo.openwire.command.WireFormatInfo;
import org.apache.geronimo.openwire.command.Command;
+import org.apache.geronimo.openwire.command.WireFormatInfo;
import org.apache.geronimo.openwire.wireformat.WireFormat;
import edu.emory.mathcs.backport.java.util.concurrent.CountDownLatch;
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgent.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgent.java?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgent.java (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgent.java Sun Sep 17 05:35:45 2006
@@ -0,0 +1,61 @@
+/**
+ *
+ * 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.geronimo.openwire.transport.discovery;
+
+import java.io.IOException;
+
+import org.apache.geronimo.openwire.Service;
+import org.apache.geronimo.openwire.command.DiscoveryEvent;
+
+/**
+ * An agent used to discover other instances of a service.
+ *
+ * We typically use a discovery agent to auto-discover JMS clients and JMS brokers on a network
+ *
+ * @version $Revision$
+ */
+public interface DiscoveryAgent extends Service {
+
+ /**
+ * Sets the discovery listener
+ * @param listener
+ */
+ public void setDiscoveryListener(DiscoveryListener listener);
+
+ /**
+ * register a service
+ * @param name
+ * @param details
+ * @throws JMSException
+ */
+ void registerService(String name) throws IOException;
+
+ /**
+ * A process actively using a service may see it go down before the DiscoveryAgent notices the
+ * service's failure. That process can use this method to notify the DiscoveryAgent of the failure
+ * so that other listeners of this DiscoveryAgent can also be made aware of the failure.
+ */
+ void serviceFailed(DiscoveryEvent event) throws IOException;
+
+ String getGroup();
+
+ void setGroup(String group);
+
+ public void setBrokerName(String brokerName);
+
+}
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgent.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgent.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgent.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgentFactory.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgentFactory.java?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgentFactory.java (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgentFactory.java Sun Sep 17 05:35:45 2006
@@ -0,0 +1,78 @@
+/**
+ *
+ * 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.geronimo.openwire.transport.discovery;
+
+import java.io.IOException;
+import java.net.URI;
+
+import org.apache.geronimo.openwire.util.FactoryFinder;
+import org.apache.geronimo.openwire.util.IOExceptionSupport;
+
+import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
+
+public abstract class DiscoveryAgentFactory {
+
+ static final private FactoryFinder discoveryAgentFinder = new FactoryFinder("META-INF/services/org/apache/activemq/transport/discoveryagent/");
+ static final private ConcurrentHashMap discoveryAgentFactorys = new ConcurrentHashMap();
+
+ /**
+ * @param uri
+ * @return
+ * @throws IOException
+ */
+ private static DiscoveryAgentFactory findDiscoveryAgentFactory(URI uri) throws IOException {
+ String scheme = uri.getScheme();
+ if( scheme == null )
+ throw new IOException("DiscoveryAgent scheme not specified: [" + uri + "]");
+ DiscoveryAgentFactory daf = (DiscoveryAgentFactory) discoveryAgentFactorys.get(scheme);
+ if (daf == null) {
+ // Try to load if from a META-INF property.
+ try {
+ daf = (DiscoveryAgentFactory) discoveryAgentFinder.newInstance(scheme);
+ discoveryAgentFactorys.put(scheme, daf);
+ }
+ catch (Throwable e) {
+ throw IOExceptionSupport.create("DiscoveryAgent scheme NOT recognized: [" + scheme + "]", e);
+ }
+ }
+ return daf;
+ }
+
+ public static DiscoveryAgent createDiscoveryAgent(URI uri) throws IOException {
+ DiscoveryAgentFactory tf = findDiscoveryAgentFactory(uri);
+ return tf.doCreateDiscoveryAgent(uri);
+
+ }
+
+ abstract protected DiscoveryAgent doCreateDiscoveryAgent(URI uri) throws IOException;
+// {
+// try {
+// String type = ( uri.getScheme() == null ) ? uri.getPath() : uri.getScheme();
+// DiscoveryAgent rc = (DiscoveryAgent) discoveryAgentFinder.newInstance(type);
+// Map options = URISupport.parseParamters(uri);
+// IntrospectionSupport.setProperties(rc, options);
+// if( rc.getClass() == SimpleDiscoveryAgent.class ) {
+// CompositeData data = URISupport.parseComposite(uri);
+// ((SimpleDiscoveryAgent)rc).setServices(data.getComponents());
+// }
+// return rc;
+// } catch (Throwable e) {
+// throw IOExceptionSupport.create("Could not create discovery agent: "+uri, e);
+// }
+// }
+}
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgentFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgentFactory.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryAgentFactory.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryListener.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryListener.java?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryListener.java (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryListener.java Sun Sep 17 05:35:45 2006
@@ -0,0 +1,32 @@
+/**
+ *
+ * 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.geronimo.openwire.transport.discovery;
+
+import org.apache.geronimo.openwire.command.DiscoveryEvent;
+
+/**
+ * A listener of services being added or removed from a network
+ *
+ * @version $Revision$
+ */
+public interface DiscoveryListener {
+
+ public void onServiceAdd(DiscoveryEvent event);
+ public void onServiceRemove(DiscoveryEvent event);
+
+}
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryListener.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryListener.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryListener.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransport.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransport.java?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransport.java (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransport.java Sun Sep 17 05:35:45 2006
@@ -0,0 +1,98 @@
+/**
+ *
+ * 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.geronimo.openwire.transport.discovery;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.geronimo.openwire.command.DiscoveryEvent;
+import org.apache.geronimo.openwire.transport.CompositeTransport;
+import org.apache.geronimo.openwire.transport.TransportFilter;
+import org.apache.geronimo.openwire.util.ServiceStopper;
+
+import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * A {@link ReliableTransportChannel} which uses a {@link DiscoveryAgent} to
+ * discover remote broker instances and dynamically connect to them.
+ *
+ * @version $Revision$
+ */
+public class DiscoveryTransport extends TransportFilter implements DiscoveryListener {
+
+ private static final Log log = LogFactory.getLog(DiscoveryTransport.class);
+
+ private final CompositeTransport next;
+ private DiscoveryAgent discoveryAgent;
+ private final ConcurrentHashMap serviceURIs = new ConcurrentHashMap();
+
+ public DiscoveryTransport(CompositeTransport next) {
+ super(next);
+ this.next = next;
+ }
+
+ public void start() throws Exception {
+ if (discoveryAgent == null) {
+ throw new IllegalStateException("discoveryAgent not configured");
+ }
+
+ // lets pass into the agent the broker name and connection details
+ discoveryAgent.setDiscoveryListener(this);
+ discoveryAgent.start();
+ next.start();
+ }
+
+ public void stop() throws Exception {
+ ServiceStopper ss = new ServiceStopper();
+ ss.stop(discoveryAgent);
+ ss.stop(next);
+ ss.throwFirstException();
+ }
+
+ public void onServiceAdd(DiscoveryEvent event) {
+ String url = event.getServiceName();
+ if (url != null) {
+ try {
+ URI uri = new URI(url);
+ serviceURIs.put(event.getServiceName(), uri);
+ log.info("Adding new broker connection URL: " + uri );
+ next.add(new URI[]{uri});
+ } catch (URISyntaxException e) {
+ log.warn("Could not connect to remote URI: " + url + " due to bad URI syntax: " + e, e);
+ }
+ }
+ }
+
+ public void onServiceRemove(DiscoveryEvent event) {
+ URI uri = (URI) serviceURIs.get(event.getServiceName());
+ if (uri != null) {
+ next.remove(new URI[]{uri});
+ }
+ }
+
+ public DiscoveryAgent getDiscoveryAgent() {
+ return discoveryAgent;
+ }
+
+ public void setDiscoveryAgent(DiscoveryAgent discoveryAgent) {
+ this.discoveryAgent = discoveryAgent;
+ }
+
+}
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransport.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransport.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransport.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransportFactory.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransportFactory.java?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransportFactory.java (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransportFactory.java Sun Sep 17 05:35:45 2006
@@ -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.geronimo.openwire.transport.discovery;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.geronimo.openwire.transport.Transport;
+import org.apache.geronimo.openwire.transport.TransportServer;
+import org.apache.geronimo.openwire.transport.failover.FailoverTransportFactory;
+import org.apache.geronimo.openwire.util.IntrospectionSupport;
+import org.apache.geronimo.openwire.util.URISupport.CompositeData;
+
+/**
+ * @version $Revision$
+ */
+public class DiscoveryTransportFactory extends FailoverTransportFactory {
+
+ public Transport createTransport(CompositeData compositData) throws IOException {
+ Map parameters = new HashMap(compositData.getParameters());
+ DiscoveryTransport transport = new DiscoveryTransport(createTransport(parameters));
+
+ DiscoveryAgent discoveryAgent = DiscoveryAgentFactory.createDiscoveryAgent(compositData.getComponents()[0]);
+ transport.setDiscoveryAgent(discoveryAgent);
+ IntrospectionSupport.setProperties(transport,parameters);
+
+ return transport;
+ }
+
+ public TransportServer doBind(String brokerId,URI location) throws IOException{
+ throw new IOException("Invalid server URI: "+location);
+ }
+
+}
\ No newline at end of file
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransportFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransportFactory.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/DiscoveryTransportFactory.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgent.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgent.java?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgent.java (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgent.java Sun Sep 17 05:35:45 2006
@@ -0,0 +1,405 @@
+/**
+ *
+ * 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.geronimo.openwire.transport.discovery.multicast;
+
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.MulticastSocket;
+import java.net.SocketAddress;
+import java.net.SocketTimeoutException;
+import java.net.URI;
+import java.util.Iterator;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.geronimo.openwire.command.DiscoveryEvent;
+import org.apache.geronimo.openwire.transport.discovery.DiscoveryAgent;
+import org.apache.geronimo.openwire.transport.discovery.DiscoveryListener;
+
+import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
+import edu.emory.mathcs.backport.java.util.concurrent.Executor;
+import edu.emory.mathcs.backport.java.util.concurrent.LinkedBlockingQueue;
+import edu.emory.mathcs.backport.java.util.concurrent.ThreadFactory;
+import edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor;
+import edu.emory.mathcs.backport.java.util.concurrent.TimeUnit;
+import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
+import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicLong;
+/**
+ * A {@link DiscoveryAgent} using <a href="http://www.zeroconf.org/">Zeroconf</a> via the <a
+ * href="http://jmdns.sf.net/">jmDNS</a> library
+ *
+ * @version $Revision$
+ */
+public class MulticastDiscoveryAgent implements DiscoveryAgent,Runnable{
+ private static final Log log=LogFactory.getLog(MulticastDiscoveryAgent.class);
+ public static final String DEFAULT_DISCOVERY_URI_STRING="multicast://224.1.2.3:6155";
+ private static final String TYPE_SUFFIX="ActiveMQ-4.";
+ private static final String ALIVE="alive.";
+ private static final String DEAD="dead.";
+ private static final String DELIMITER = "%";
+ private static final int BUFF_SIZE=8192;
+ private static final int DEFAULT_IDLE_TIME=500;
+ private static final int HEARTBEAT_MISS_BEFORE_DEATH=4;
+ private int timeToLive=1;
+ private boolean loopBackMode=false;
+ private Map services=new ConcurrentHashMap();
+ private Map brokers = new ConcurrentHashMap();
+ private String group="default";
+ private String brokerName;
+ private URI discoveryURI;
+ private InetAddress inetAddress;
+ private SocketAddress sockAddress;
+ private DiscoveryListener discoveryListener;
+ private String selfService;
+ private MulticastSocket mcast;
+ private Thread runner;
+ private long keepAliveInterval=DEFAULT_IDLE_TIME;
+ private long lastAdvertizeTime=0;
+ private AtomicBoolean started=new AtomicBoolean(false);
+ private boolean reportAdvertizeFailed=true;
+
+ private final Executor executor = new ThreadPoolExecutor(1, 1, 30, TimeUnit.SECONDS, new LinkedBlockingQueue(), new ThreadFactory() {
+ public Thread newThread(Runnable runable) {
+ Thread t = new Thread(runable, "Multicast Discovery Agent Notifier");
+ t.setDaemon(true);
+ return t;
+ }
+ });
+
+ /**
+ * Set the discovery listener
+ *
+ * @param listener
+ */
+ public void setDiscoveryListener(DiscoveryListener listener){
+ this.discoveryListener=listener;
+ }
+
+ /**
+ * register a service
+ */
+ public void registerService(String name) throws IOException{
+ this.selfService=name;
+ if (started.get()){
+ doAdvertizeSelf();
+ }
+ }
+
+ /**
+ * Get the group used for discovery
+ *
+ * @return the group
+ */
+ public String getGroup(){
+ return group;
+ }
+
+ /**
+ * Set the group for discovery
+ *
+ * @param group
+ */
+ public void setGroup(String group){
+ this.group=group;
+ }
+
+ /**
+ * @return Returns the brokerName.
+ */
+ public String getBrokerName(){
+ return brokerName;
+ }
+
+ /**
+ * @param brokerName The brokerName to set.
+ */
+ public void setBrokerName(String brokerName){
+ if (brokerName != null){
+ brokerName = brokerName.replace('.','-');
+ brokerName = brokerName.replace(':','-');
+ brokerName = brokerName.replace('%','-');
+ this.brokerName=brokerName;
+ }
+ }
+
+ /**
+ * @return Returns the loopBackMode.
+ */
+ public boolean isLoopBackMode(){
+ return loopBackMode;
+ }
+
+ /**
+ * @param loopBackMode
+ * The loopBackMode to set.
+ */
+ public void setLoopBackMode(boolean loopBackMode){
+ this.loopBackMode=loopBackMode;
+ }
+
+ /**
+ * @return Returns the timeToLive.
+ */
+ public int getTimeToLive(){
+ return timeToLive;
+ }
+
+ /**
+ * @param timeToLive
+ * The timeToLive to set.
+ */
+ public void setTimeToLive(int timeToLive){
+ this.timeToLive=timeToLive;
+ }
+
+ /**
+ * @return the discoveryURI
+ */
+ public URI getDiscoveryURI(){
+ return discoveryURI;
+ }
+
+ /**
+ * Set the discoveryURI
+ *
+ * @param discoveryURI
+ */
+ public void setDiscoveryURI(URI discoveryURI){
+ this.discoveryURI=discoveryURI;
+ }
+
+ public long getKeepAliveInterval(){
+ return keepAliveInterval;
+ }
+
+ public void setKeepAliveInterval(long keepAliveInterval){
+ this.keepAliveInterval=keepAliveInterval;
+ }
+
+ /**
+ * start the discovery agent
+ *
+ * @throws Exception
+ */
+ public void start() throws Exception{
+ if(started.compareAndSet(false,true)){
+ if(group==null|| group.length()==0){
+ throw new IOException("You must specify a group to discover");
+ }
+ if (brokerName == null || brokerName.length()==0){
+ log.warn("brokerName not set");
+ }
+ String type=getType();
+ if(!type.endsWith(".")){
+ log.warn("The type '"+type+"' should end with '.' to be a valid Discovery type");
+ type+=".";
+ }
+ if(discoveryURI==null){
+ discoveryURI=new URI(DEFAULT_DISCOVERY_URI_STRING);
+ }
+ this.inetAddress=InetAddress.getByName(discoveryURI.getHost());
+ this.sockAddress=new InetSocketAddress(this.inetAddress,discoveryURI.getPort());
+ mcast=new MulticastSocket(discoveryURI.getPort());
+ mcast.setLoopbackMode(loopBackMode);
+ mcast.setTimeToLive(getTimeToLive());
+ mcast.joinGroup(inetAddress);
+ mcast.setSoTimeout((int) keepAliveInterval);
+ runner=new Thread(this);
+ runner.setName("MulticastDiscovery: "+selfService);
+ runner.setDaemon(true);
+ runner.start();
+ doAdvertizeSelf();
+ }
+ }
+
+ /**
+ * stop the channel
+ *
+ * @throws Exception
+ */
+ public void stop() throws Exception{
+ if(started.compareAndSet(true,false)){
+ doAdvertizeSelf();
+ mcast.close();
+ }
+ }
+
+ public String getType(){
+ return group+"."+TYPE_SUFFIX;
+ }
+
+ public void run(){
+ byte[] buf=new byte[BUFF_SIZE];
+ DatagramPacket packet=new DatagramPacket(buf,0,buf.length);
+ while(started.get()){
+ doTimeKeepingServices();
+ try{
+ mcast.receive(packet);
+ if(packet.getLength()>0){
+ String str=new String(packet.getData(),packet.getOffset(),packet.getLength());
+ processData(str);
+ }
+ } catch(SocketTimeoutException se){
+ // ignore
+ } catch(IOException e){
+ if( started.get() ) {
+ log.error("failed to process packet: "+e);
+ }
+ }
+ }
+ }
+
+ private void processData(String str){
+ if (discoveryListener != null){
+ if(str.startsWith(getType())){
+ String payload=str.substring(getType().length());
+ if(payload.startsWith(ALIVE)){
+ String brokerName=getBrokerName(payload.substring(ALIVE.length()));
+ String service=payload.substring(ALIVE.length()+brokerName.length()+2);
+ if(!brokerName.equals(this.brokerName)){
+ processAlive(brokerName,service);
+ }
+ }else{
+ String brokerName=getBrokerName(payload.substring(DEAD.length()));
+ String service=payload.substring(DEAD.length()+brokerName.length()+2);
+ if(!brokerName.equals(this.brokerName)){
+ processDead(brokerName,service);
+ }
+ }
+ }
+ }
+ }
+
+ private void doTimeKeepingServices(){
+ if(started.get()){
+ long currentTime=System.currentTimeMillis();
+ if((currentTime-keepAliveInterval)>lastAdvertizeTime){
+ doAdvertizeSelf();
+ lastAdvertizeTime = currentTime;
+ }
+ doExpireOldServices();
+ }
+ }
+
+ private void doAdvertizeSelf(){
+ if(selfService!=null ){
+ String payload=getType();
+ payload+=started.get()?ALIVE:DEAD;
+ payload+=DELIMITER+brokerName+DELIMITER;
+ payload+=selfService;
+ try{
+ byte[] data=payload.getBytes();
+ DatagramPacket packet=new DatagramPacket(data,0,data.length,sockAddress);
+ mcast.send(packet);
+ } catch(IOException e) {
+ // If a send fails, chances are all subsequent sends will fail too.. No need to keep reporting the
+ // same error over and over.
+ if( reportAdvertizeFailed ) {
+ reportAdvertizeFailed=false;
+ log.error("Failed to advertise our service: "+payload,e);
+ if( "Operation not permitted".equals(e.getMessage()) ) {
+ log.error("The 'Operation not permitted' error has been know to be caused by improper firewall/network setup. Please make sure that the OS is properly configured to allow multicast traffic over: "+mcast.getLocalAddress());
+ }
+ }
+ }
+ }
+ }
+
+ private void processAlive(String brokerName,String service){
+ if(selfService == null || !service.equals(selfService)){
+ AtomicLong lastKeepAlive=(AtomicLong) services.get(service);
+ if(lastKeepAlive==null){
+ brokers.put(service, brokerName);
+ if(discoveryListener!=null){
+ final DiscoveryEvent event=new DiscoveryEvent(service);
+ event.setBrokerName(brokerName);
+
+ // Have the listener process the event async so that
+ // he does not block this thread since we are doing time sensitive
+ // processing of events.
+ executor.execute(new Runnable() {
+ public void run() {
+ DiscoveryListener discoveryListener = MulticastDiscoveryAgent.this.discoveryListener;
+ if(discoveryListener!=null){
+ discoveryListener.onServiceAdd(event);
+ }
+ }
+ });
+ }
+ lastKeepAlive=new AtomicLong(System.currentTimeMillis());
+ services.put(service,lastKeepAlive);
+ doAdvertizeSelf();
+
+ }
+ lastKeepAlive.set(System.currentTimeMillis());
+ }
+ }
+
+ private void processDead(String brokerName,String service){
+ if(!service.equals(selfService)){
+ if(services.remove(service)!=null){
+ brokers.remove(service);
+ if(discoveryListener!=null){
+ final DiscoveryEvent event=new DiscoveryEvent(service);
+ event.setBrokerName(brokerName);
+
+ // Have the listener process the event async so that
+ // he does not block this thread since we are doing time sensitive
+ // processing of events.
+ executor.execute(new Runnable() {
+ public void run() {
+ DiscoveryListener discoveryListener = MulticastDiscoveryAgent.this.discoveryListener;
+ if(discoveryListener!=null){
+ discoveryListener.onServiceRemove(event);
+ }
+ }
+ });
+ }
+ }
+ }
+ }
+
+ private void doExpireOldServices(){
+ long expireTime=System.currentTimeMillis()-(keepAliveInterval*HEARTBEAT_MISS_BEFORE_DEATH);
+ for(Iterator i=services.entrySet().iterator();i.hasNext();){
+ Map.Entry entry=(Map.Entry) i.next();
+ AtomicLong lastHeartBeat=(AtomicLong) entry.getValue();
+ if(lastHeartBeat.get()<expireTime){
+ String brokerName = (String)brokers.get(entry.getKey());
+ processDead(brokerName,entry.getKey().toString());
+ }
+ }
+ }
+
+ private String getBrokerName(String str){
+ String result = null;
+ int start = str.indexOf(DELIMITER);
+ if (start >= 0 ){
+ int end = str.indexOf(DELIMITER,start+1);
+ result=str.substring(start+1, end);
+ }
+ return result;
+ }
+
+ public void serviceFailed(DiscoveryEvent event) throws IOException {
+ processDead(event.getBrokerName(), event.getServiceName());
+ }
+}
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgent.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgent.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgent.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgentFactory.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgentFactory.java?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgentFactory.java (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgentFactory.java Sun Sep 17 05:35:45 2006
@@ -0,0 +1,45 @@
+/**
+ *
+ * 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.geronimo.openwire.transport.discovery.multicast;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+
+import org.apache.geronimo.openwire.transport.discovery.DiscoveryAgent;
+import org.apache.geronimo.openwire.transport.discovery.DiscoveryAgentFactory;
+import org.apache.geronimo.openwire.util.IOExceptionSupport;
+import org.apache.geronimo.openwire.util.IntrospectionSupport;
+import org.apache.geronimo.openwire.util.URISupport;
+
+public class MulticastDiscoveryAgentFactory extends DiscoveryAgentFactory {
+
+ protected DiscoveryAgent doCreateDiscoveryAgent(URI uri) throws IOException {
+ try {
+
+ Map options = URISupport.parseParamters(uri);
+ MulticastDiscoveryAgent rc = new MulticastDiscoveryAgent();
+ rc.setGroup(uri.getHost());
+ IntrospectionSupport.setProperties(rc, options);
+ return rc;
+
+ } catch (Throwable e) {
+ throw IOExceptionSupport.create("Could not create discovery agent: " + uri, e);
+ }
+ }
+}
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgentFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgentFactory.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/multicast/MulticastDiscoveryAgentFactory.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/package.html
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/package.html?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/package.html (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/package.html Sun Sep 17 05:35:45 2006
@@ -0,0 +1,25 @@
+<!--
+ 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.
+-->
+<html>
+<head>
+</head>
+<body>
+
+Discovery mechanism to discover brokers and clients.
+
+</body>
+</html>
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/package.html
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/package.html
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/package.html
------------------------------------------------------------------------------
svn:mime-type = text/html
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgent.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgent.java?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgent.java (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgent.java Sun Sep 17 05:35:45 2006
@@ -0,0 +1,222 @@
+/**
+ *
+ * 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.geronimo.openwire.transport.discovery.simple;
+
+import java.io.IOException;
+import java.net.URI;
+
+import org.apache.geronimo.openwire.command.DiscoveryEvent;
+import org.apache.geronimo.openwire.transport.discovery.DiscoveryAgent;
+import org.apache.geronimo.openwire.transport.discovery.DiscoveryListener;
+
+import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A simple DiscoveryAgent that allows static configuration of the discovered services.
+ *
+ * @version $Revision$
+ */
+public class SimpleDiscoveryAgent implements DiscoveryAgent {
+
+ private long initialReconnectDelay = 1000*5;
+ private long maxReconnectDelay = 1000 * 30;
+ private long backOffMultiplier = 2;
+ private boolean useExponentialBackOff = false;
+ private int maxReconnectAttempts;
+ private final Object sleepMutex = new Object();
+ private long minConnectTime = 500;
+
+ private DiscoveryListener listener;
+ String services[] = new String[] {};
+ String group = "DEFAULT";
+ private final AtomicBoolean running = new AtomicBoolean(false);
+
+ class SimpleDiscoveryEvent extends DiscoveryEvent {
+
+ private int connectFailures;
+ private long reconnectDelay = initialReconnectDelay;
+ private long connectTime = System.currentTimeMillis();
+ private AtomicBoolean failed = new AtomicBoolean(false);
+
+ public SimpleDiscoveryEvent(String service) {
+ super(service);
+ }
+
+ }
+
+ public void setDiscoveryListener(DiscoveryListener listener) {
+ this.listener = listener;
+ }
+
+ public void registerService(String name) throws IOException {
+ }
+
+ public void start() throws Exception {
+ running.set(true);
+ for (int i = 0; i < services.length; i++) {
+ listener.onServiceAdd(new SimpleDiscoveryEvent(services[i]));
+ }
+ }
+
+ public void stop() throws Exception {
+ running.set(false);
+ synchronized(sleepMutex) {
+ sleepMutex.notifyAll();
+ }
+ }
+
+ public String[] getServices() {
+ return services;
+ }
+
+ public void setServices(String services) {
+ this.services = services.split(",");
+ }
+
+ public void setServices(String services[]) {
+ this.services = services;
+ }
+
+ public void setServices(URI services[]) {
+ this.services = new String[services.length];
+ for (int i = 0; i < services.length; i++) {
+ this.services[i] = services[i].toString();
+ }
+ }
+
+ public String getGroup() {
+ return group;
+ }
+
+ public void setGroup(String group) {
+ this.group = group;
+ }
+
+ public void setBrokerName(String brokerName) {
+ }
+
+ public void serviceFailed(DiscoveryEvent devent) throws IOException {
+
+ final SimpleDiscoveryEvent event = (SimpleDiscoveryEvent) devent;
+ if( event.failed.compareAndSet(false, true) ) {
+
+ listener.onServiceRemove(event);
+ Thread thread = new Thread() {
+ public void run() {
+
+
+ // We detect a failed connection attempt because the service fails right
+ // away.
+ if( event.connectTime + minConnectTime > System.currentTimeMillis() ) {
+
+ event.connectFailures++;
+
+ if( maxReconnectAttempts>0 && event.connectFailures >= maxReconnectAttempts ) {
+ // Don' try to re-connect
+ return;
+ }
+
+ synchronized(sleepMutex){
+ try{
+ if( !running.get() )
+ return;
+
+ sleepMutex.wait(event.reconnectDelay);
+ }catch(InterruptedException ie){
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+
+ if (!useExponentialBackOff) {
+ event.reconnectDelay = initialReconnectDelay;
+ } else {
+ // Exponential increment of reconnect delay.
+ event.reconnectDelay*=backOffMultiplier;
+ if(event.reconnectDelay>maxReconnectDelay)
+ event.reconnectDelay=maxReconnectDelay;
+ }
+
+ } else {
+ event.connectFailures = 0;
+ event.reconnectDelay = initialReconnectDelay;
+ }
+
+ if( !running.get() )
+ return;
+
+ event.connectTime = System.currentTimeMillis();
+ event.failed.set(false);
+
+ listener.onServiceAdd(event);
+ }
+ };
+ thread.setDaemon(true);
+ thread.start();
+ }
+ }
+
+ public long getBackOffMultiplier() {
+ return backOffMultiplier;
+ }
+
+ public void setBackOffMultiplier(long backOffMultiplier) {
+ this.backOffMultiplier = backOffMultiplier;
+ }
+
+ public long getInitialReconnectDelay() {
+ return initialReconnectDelay;
+ }
+
+ public void setInitialReconnectDelay(long initialReconnectDelay) {
+ this.initialReconnectDelay = initialReconnectDelay;
+ }
+
+ 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 long getMinConnectTime() {
+ return minConnectTime;
+ }
+
+ public void setMinConnectTime(long minConnectTime) {
+ this.minConnectTime = minConnectTime;
+ }
+
+ public boolean isUseExponentialBackOff() {
+ return useExponentialBackOff;
+ }
+
+ public void setUseExponentialBackOff(boolean useExponentialBackOff) {
+ this.useExponentialBackOff = useExponentialBackOff;
+ }
+
+}
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgent.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgent.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgent.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgentFactory.java
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgentFactory.java?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgentFactory.java (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgentFactory.java Sun Sep 17 05:35:45 2006
@@ -0,0 +1,50 @@
+/**
+ *
+ * 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.geronimo.openwire.transport.discovery.simple;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Map;
+
+import org.apache.geronimo.openwire.transport.discovery.DiscoveryAgent;
+import org.apache.geronimo.openwire.transport.discovery.DiscoveryAgentFactory;
+import org.apache.geronimo.openwire.util.IOExceptionSupport;
+import org.apache.geronimo.openwire.util.IntrospectionSupport;
+import org.apache.geronimo.openwire.util.URISupport;
+import org.apache.geronimo.openwire.util.URISupport.CompositeData;
+
+public class SimpleDiscoveryAgentFactory extends DiscoveryAgentFactory {
+
+ protected DiscoveryAgent doCreateDiscoveryAgent(URI uri) throws IOException {
+ try {
+
+ CompositeData data = URISupport.parseComposite(uri);
+ Map options = URISupport.parseParamters(uri);
+
+ SimpleDiscoveryAgent rc = new SimpleDiscoveryAgent();
+ rc.setGroup(uri.getHost());
+ IntrospectionSupport.setProperties(rc, options);
+ rc.setServices(data.getComponents());
+
+ return rc;
+
+ } catch (Throwable e) {
+ throw IOExceptionSupport.create("Could not create discovery agent: " + uri, e);
+ }
+ }
+}
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgentFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgentFactory.java
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/SimpleDiscoveryAgentFactory.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/package.html
URL: http://svn.apache.org/viewvc/geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/package.html?view=auto&rev=447058
==============================================================================
--- geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/package.html (added)
+++ geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/package.html Sun Sep 17 05:35:45 2006
@@ -0,0 +1,25 @@
+<!--
+ 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.
+-->
+<html>
+<head>
+</head>
+<body>
+
+ Simple discovery implementation using a static list
+
+</body>
+</html>
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/package.html
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/package.html
------------------------------------------------------------------------------
svn:keywords = Date Revision
Propchange: geronimo/sandbox/gcache/openwire/src/main/java/org/apache/geronimo/openwire/transport/discovery/simple/package.html
------------------------------------------------------------------------------
svn:mime-type = text/html