You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tuscany.apache.org by me...@apache.org on 2007/01/24 08:31:04 UTC
svn commit: r499293 - in
/incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta:
JxtaDiscoveryService.java PeerListener.java
Author: meerajk
Date: Tue Jan 23 23:31:04 2007
New Revision: 499293
URL: http://svn.apache.org/viewvc?view=rev&rev=499293
Log:
Added peer listener.
Added:
incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/PeerListener.java (with props)
Modified:
incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/JxtaDiscoveryService.java
Modified: incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/JxtaDiscoveryService.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/JxtaDiscoveryService.java?view=diff&rev=499293&r1=499292&r2=499293
==============================================================================
--- incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/JxtaDiscoveryService.java (original)
+++ incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/JxtaDiscoveryService.java Tue Jan 23 23:31:04 2007
@@ -25,6 +25,7 @@
import javax.xml.stream.XMLStreamReader;
import net.jxta.credential.AuthenticationCredential;
+import net.jxta.discovery.DiscoveryService;
import net.jxta.exception.PeerGroupException;
import net.jxta.membership.Authenticator;
import net.jxta.membership.MembershipService;
@@ -33,10 +34,11 @@
import net.jxta.platform.NetworkConfigurator;
import net.jxta.protocol.ModuleImplAdvertisement;
-import org.apache.tuscany.host.RuntimeInfo;
import org.apache.tuscany.spi.annotation.Autowire;
import org.apache.tuscany.spi.services.discovery.AbstractDiscoveryService;
+import org.apache.tuscany.spi.services.work.WorkScheduler;
import org.omg.CORBA.Any;
+import org.osoa.sca.annotations.Property;
/**
* Discovery service implemented using Apple bonjour.
@@ -45,12 +47,24 @@
*
*/
public class JxtaDiscoveryService extends AbstractDiscoveryService {
+
+ /** Default discovery interval. */
+ private static long DEFAULT_INTERVAL = 2000L;
- /** Pipe receiver. */
- private PipeReceiver pipeReceiver;
+ /** Peer listener. */
+ private PeerListener peerListener;
/** Network platform configurator. */
private NetworkConfigurator configurator;
+
+ /** Work scheduler. */
+ private WorkScheduler workScheduler;
+
+ /** Interval for sending discivery messages .*/
+ private long interval = DEFAULT_INTERVAL;
+
+ /** Domain peer group. */
+ private PeerGroup domainGroup;
/**
* Adds a network configurator for this service.
@@ -62,6 +76,24 @@
}
/**
+ * Adds a work scheduler for runningbackground discovery operations.
+ * @param workScheduler Work scheduler.
+ */
+ @Autowire
+ public void setWorkScheduler(WorkScheduler workScheduler) {
+ this.workScheduler = workScheduler;
+ }
+
+ /**
+ * Sets the interval at which discovery messages are sent.
+ * @param interval Interval at which discovery messages are sent.
+ */
+ @Property
+ public void setInterval(long interval) {
+ this.interval = interval;
+ }
+
+ /**
* Starts the discovery service.
* @throws Any unexpected JXTA exception to bubble up the call stack.
*/
@@ -72,15 +104,9 @@
// Configure the platform
configure();
+
String domain = getRuntimeInfo().getDomain().toString();
-
- PeerGroup domainGroup = createAndJoinDomainGroup(domain);
- pipeReceiver = PipeReceiver.newInstance(this, domainGroup);
-
- RuntimeInfo runtimeInfo = getRuntimeInfo();
- String runtimeId = runtimeInfo.getRuntimeId();
-
- pipeReceiver.start(domain, runtimeId);
+ createAndJoinDomainGroup(domain);
} catch (PeerGroupException ex) {
throw new JxtaException(ex);
@@ -98,8 +124,19 @@
* @param runtimeId Runtime id of recipient.
* @param content Message content.
*/
- public void sendMessage(String runtimeId, XMLStreamReader content) {
- throw new UnsupportedOperationException();
+ public void sendMessage(final String runtimeId, final XMLStreamReader content) {
+
+ if(runtimeId == null) {
+ throw new IllegalArgumentException("Runtime id is null");
+ }
+ if(content == null) {
+ throw new IllegalArgumentException("Content id is null");
+ }
+
+ if(peerListener.isRuntimeAlive(runtimeId)) {
+ // Send the message
+ }
+
}
/**
@@ -107,7 +144,7 @@
*/
@Override
protected void onStop() {
- pipeReceiver.stop();
+ peerListener.stop();
}
/**
@@ -141,13 +178,15 @@
* @return Domain peer group.
* @throws Exception In case of unexpected JXTA exceptions.
*/
- private PeerGroup createAndJoinDomainGroup(String domain) throws Exception {
+ private void createAndJoinDomainGroup(String domain) throws Exception {
PeerGroup netGroup = new NetPeerGroupFactory().getInterface();
ModuleImplAdvertisement implAdv = netGroup.getAllPurposePeerGroupImplAdvertisement();
- PeerGroup domainGroup = netGroup.newGroup(null, implAdv, domain, "Tuscany domain group");
- domainGroup.getDiscoveryService().remotePublish(domainGroup.getPeerGroupAdvertisement());
+ domainGroup = netGroup.newGroup(null, implAdv, domain, "Tuscany domain group");
+
+ final DiscoveryService discoveryService = netGroup.getDiscoveryService();
+ discoveryService.remotePublish(netGroup.getPeerAdvertisement());
AuthenticationCredential authCred = new AuthenticationCredential(domainGroup, null, null);
MembershipService membership = domainGroup.getMembershipService();
@@ -155,10 +194,13 @@
if (auth.isReadyForJoin()){
membership.join(auth);
+ System.err.println("Joined" + domainGroup.getPeerGroupName());
} else {
throw new JxtaException("Unable to join domain group");
}
- return domainGroup;
+
+ peerListener = new PeerListener(discoveryService, interval);
+ workScheduler.scheduleWork(peerListener);
}
Added: incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/PeerListener.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/PeerListener.java?view=auto&rev=499293
==============================================================================
--- incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/PeerListener.java (added)
+++ incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/PeerListener.java Tue Jan 23 23:31:04 2007
@@ -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.tuscany.service.discovery.jxta;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import net.jxta.discovery.DiscoveryEvent;
+import net.jxta.discovery.DiscoveryListener;
+import net.jxta.discovery.DiscoveryService;
+
+/**
+ * Listener that keeps track of peers in the same peer group.
+ *
+ * @version $Revision$ $Date$
+ *
+ */
+public class PeerListener implements Runnable, DiscoveryListener {
+
+ /** Discovery service to use. */
+ private DiscoveryService discoveryService;
+
+ /** Interval for sending discivery messages. */
+ private long interval;
+
+ /** Liveness indicator. */
+ private AtomicBoolean live = new AtomicBoolean();
+
+ /** Available peers. */
+ private List<String> availablePeers = new LinkedList<String>();
+
+ /**
+ * Initializes the JXTA discovery service.
+ * @param discoveryService JXTA discovery service.
+ */
+ public PeerListener(DiscoveryService discoveryService, long interval) {
+ this.discoveryService = discoveryService;
+ }
+
+ /**
+ * Sends discovery messages for peer advertisements.
+ */
+ public void run() {
+
+ discoveryService.addDiscoveryListener(this);
+ live.set(true);
+
+ while(live.get()) {
+ discoveryService.getRemoteAdvertisements(null, DiscoveryService.PEER, null, null, 5);
+ try {
+ Thread.sleep(interval);
+ } catch (InterruptedException e) {
+ return;
+ }
+ }
+
+ }
+
+ /**
+ * Returns the available peers participating in the domain.
+ * @return True if the specified runtime is alive.
+ */
+ public synchronized boolean isRuntimeAlive(String runtimeId) {
+ return availablePeers.contains(runtimeId);
+ }
+
+ /**
+ * Receives asynchronous discovery responses.
+ */
+ public synchronized void discoveryEvent(DiscoveryEvent discoveryEvent) {
+ // TODO Update the list of available peers
+ }
+
+ /**
+ * Stops the pipe listener.
+ */
+ public void stop() {
+ live.set(false);
+ }
+
+}
Propchange: incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/PeerListener.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/tuscany/java/sca/services/discovery/jxta/src/main/java/org/apache/tuscany/service/discovery/jxta/PeerListener.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
---------------------------------------------------------------------
To unsubscribe, e-mail: tuscany-commits-unsubscribe@ws.apache.org
For additional commands, e-mail: tuscany-commits-help@ws.apache.org