You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tuscany.apache.org by is...@apache.org on 2007/06/28 18:30:55 UTC
svn commit: r551614 [2/5] - in
/incubator/tuscany/java/sca/modules/binding-notification: ./ src/ src/main/
src/main/java/ src/main/java/org/ src/main/java/org/apache/
src/main/java/org/apache/tuscany/ src/main/java/org/apache/tuscany/sca/
src/main/java...
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationReferenceBindingProvider.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationReferenceBindingProvider.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationReferenceBindingProvider.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationReferenceBindingProvider.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,226 @@
+/*
+ * 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.sca.binding.notification;
+
+import java.net.URI;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import javax.servlet.ServletInputStream;
+import javax.servlet.ServletOutputStream;
+
+import org.apache.axiom.om.OMElement;
+import org.apache.tuscany.sca.binding.notification.encoding.Broker;
+import org.apache.tuscany.sca.binding.notification.encoding.ConnectionOverride;
+import org.apache.tuscany.sca.binding.notification.encoding.Constants;
+import org.apache.tuscany.sca.binding.notification.encoding.EncodingObject;
+import org.apache.tuscany.sca.binding.notification.encoding.EncodingRegistry;
+import org.apache.tuscany.sca.binding.notification.encoding.EncodingUtils;
+import org.apache.tuscany.sca.binding.notification.encoding.EndpointReference;
+import org.apache.tuscany.sca.binding.notification.encoding.ReplaceBrokerConnection;
+import org.apache.tuscany.sca.binding.notification.encoding.Subscribe;
+import org.apache.tuscany.sca.binding.notification.util.NotificationServlet;
+import org.apache.tuscany.sca.binding.notification.util.URIUtil;
+import org.apache.tuscany.sca.binding.notification.util.NotificationServlet.NotificationServletStreamHandler;
+import org.apache.tuscany.sca.http.ServletHost;
+import org.apache.tuscany.sca.interfacedef.Interface;
+import org.apache.tuscany.sca.interfacedef.InterfaceContract;
+import org.apache.tuscany.sca.interfacedef.Operation;
+import org.apache.tuscany.sca.invocation.Invoker;
+import org.apache.tuscany.sca.provider.ReferenceBindingProvider;
+import org.apache.tuscany.sca.runtime.RuntimeComponent;
+import org.apache.tuscany.sca.runtime.RuntimeComponentReference;
+
+/**
+ * The runtime representaion of the notification reference binding
+ *
+ * @version $Rev$ $Date$
+ */
+public class NotificationReferenceBindingProvider
+ implements ReferenceBindingProvider, NotificationServletStreamHandler {
+
+ private static final String producerPathBase = "/producer";
+ private NotificationReferenceBindingInvoker invoker;
+ private RuntimeComponentReference reference;
+ private NotificationBinding notificationBinding;
+ private ServletHost servletHost;
+ private NotificationTypeManager ntm;
+ private EncodingRegistry encodingRegistry;
+ private URI notificationType;
+ private URL myUrl;
+ private URL remoteNtmUrl;
+ private boolean started;
+ private NotificationBrokerManager brokerManager;
+
+ public NotificationReferenceBindingProvider(NotificationBinding notificationBinding,
+ RuntimeComponent component,
+ RuntimeComponentReference reference,
+ ServletHost servletHost,
+ NotificationTypeManager ntm,
+ EncodingRegistry encodingRegistry,
+ String httpUrl,
+ NotificationBrokerManager brokerManager) {
+ this.invoker = null;
+ this.notificationBinding = notificationBinding;
+ this.reference = reference;
+ this.servletHost = servletHost;
+ this.ntm = ntm;
+ this.encodingRegistry = encodingRegistry;
+ this.notificationType = notificationBinding.getNotificationType();
+ String ntmAddress = notificationBinding.getNtmAddress();
+ String notificationTypePath = URIUtil.getPath(notificationType);
+ try {
+ this.myUrl = new URL(httpUrl + producerPathBase + notificationTypePath);
+ this.remoteNtmUrl = null;
+ if (ntmAddress != null && notificationType != null) {
+ remoteNtmUrl = new URL(ntmAddress + notificationTypePath);
+ }
+ }
+ catch(Exception e) {
+ throw new RuntimeException(e);
+ }
+ this.started = false;
+ this.brokerManager = brokerManager;
+
+ URI uri = URI.create(component.getURI() + "/" + notificationBinding.getName());
+ notificationBinding.setURI(uri.toString());
+ Interface interfaze = reference.getInterfaceContract().getInterface();
+ interfaze.setDefaultDataBinding(OMElement.class.getName());
+ for (Operation operation : interfaze.getOperations()) {
+ operation.setNonBlocking(false);
+ }
+ }
+
+ public NotificationBinding getBinding() {
+ return notificationBinding;
+ }
+
+ public URL getURL() {
+ return myUrl;
+ }
+
+ public boolean isStarted() {
+ return started;
+ }
+
+ public Invoker createInvoker(Operation operation, boolean isCallback) {
+ if (isCallback) {
+ throw new UnsupportedOperationException();
+ }
+ if (invoker == null) {
+ invoker = new NotificationReferenceBindingInvoker(operation);
+ }
+ return invoker;
+ }
+
+ public InterfaceContract getBindingInterfaceContract() {
+ return reference.getInterfaceContract();
+ }
+
+ public void start() {
+ if (started) {
+ return;
+ }
+
+ brokerManager.referenceProviderStarted(notificationType, this, remoteNtmUrl);
+ started = true;
+ }
+
+ public void stop() {
+ }
+
+ public void deployProducer() {
+ List<URL> consumerList = new ArrayList<URL>();
+ String sequenceType;
+ try {
+ sequenceType = ntm.newProducer(notificationType, myUrl, remoteNtmUrl, consumerList);
+ } catch(Exception e) {
+ throw new RuntimeException(e);
+ }
+ if (Constants.EndConsumers.equals(sequenceType)) {
+ for (URL consumerUrl : consumerList) {
+ invoker.addSubscriberUrl(consumerUrl);
+ }
+ }
+ else if (Constants.BrokerConsumers.equals(sequenceType)) {
+ // Pick a broker consumer, for now the first one
+ URL consumerUrl = consumerList.get(0);
+ invoker.addSubscriberUrl(consumerUrl);
+ }
+
+ servletHost.addServletMapping(myUrl.toString(), new NotificationServlet(this));
+ }
+
+ public void deployBroker(String brokerID, EndpointReference brokerConsumerEPR, List<EndpointReference> consumerList) {
+ if (brokerConsumerEPR != null) {
+ invoker.addSubscriber(brokerConsumerEPR);
+ }
+ if (consumerList != null && !consumerList.isEmpty()) {
+ for (EndpointReference consumerEPR : consumerList) {
+ invoker.addSubscriber(consumerEPR);
+ }
+ }
+ invoker.setBrokerID(brokerID);
+ servletHost.addServletMapping(myUrl.toString(), new NotificationServlet(this));
+ }
+
+ public void undeployBroker(URL brokerConsumerUrl) {
+ EndpointReference brokerConsumerEpr = EncodingUtils.createEndpointReference(brokerConsumerUrl, invoker.getBrokerID());
+ ntm.removeBroker(brokerConsumerEpr, invoker.getNeighborBrokerConsumerEprs(), remoteNtmUrl);
+ invoker.removeBrokerSubscribers();
+ }
+
+ public void handle(Map<String, String> headers, ServletInputStream istream, int contentLength, ServletOutputStream ostream) {
+
+ try {
+ EncodingObject eo = EncodingUtils.decodeFromStream(encodingRegistry, istream);
+ if (eo instanceof Subscribe) {
+ Subscribe sub = (Subscribe)eo;
+ invoker.addSubscriber(sub.getConsumerReference().getReference());
+ }
+ else if (eo instanceof ConnectionOverride) {
+ ConnectionOverride co = (ConnectionOverride)eo;
+ invoker.replaceSubscribers(co.getBrokerConsumerReference().getReference());
+ }
+ else if (eo instanceof ReplaceBrokerConnection) {
+ ReplaceBrokerConnection rbc = (ReplaceBrokerConnection)eo;
+ URL removedBrokerConsumerEpr = rbc.getRemovedBroker().getReference().getEndpointAddress().getAddress();
+ if (rbc.getNeighbors() != null) {
+ int choice = rbc.getNeighbors().getBrokerSequence().size() - 1;
+ Broker chosenBroker = rbc.getNeighbors().getBrokerSequence().get(choice);
+ invoker.replaceBrokerSubscriber(removedBrokerConsumerEpr,
+ chosenBroker.getBrokerConsumerReference().getReference());
+ brokerManager.replaceConsumersBrokerConnection(notificationType,
+ chosenBroker.getBrokerProducerReference().getReference());
+ }
+ else {
+ invoker.replaceBrokerSubscriber(removedBrokerConsumerEpr, null);
+ }
+ }
+ else {
+ throw new RuntimeException("Unknown encoding object");
+ }
+ } catch(Throwable e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ }
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationServiceBindingProvider.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationServiceBindingProvider.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationServiceBindingProvider.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationServiceBindingProvider.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,313 @@
+/*
+ * 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.sca.binding.notification;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.URI;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import javax.servlet.ServletInputStream;
+import javax.servlet.ServletOutputStream;
+
+import org.apache.axiom.om.OMElement;
+import org.apache.axiom.om.impl.builder.StAXOMBuilder;
+import org.apache.tuscany.sca.binding.notification.encoding.BrokerConsumerReference;
+import org.apache.tuscany.sca.binding.notification.encoding.BrokerID;
+import org.apache.tuscany.sca.binding.notification.encoding.ConnectionOverride;
+import org.apache.tuscany.sca.binding.notification.encoding.Constants;
+import org.apache.tuscany.sca.binding.notification.encoding.ConsumerReference;
+import org.apache.tuscany.sca.binding.notification.encoding.EncodingRegistry;
+import org.apache.tuscany.sca.binding.notification.encoding.EncodingUtils;
+import org.apache.tuscany.sca.binding.notification.encoding.EndpointAddress;
+import org.apache.tuscany.sca.binding.notification.encoding.EndpointReference;
+import org.apache.tuscany.sca.binding.notification.encoding.ReferenceProperties;
+import org.apache.tuscany.sca.binding.notification.encoding.Subscribe;
+import org.apache.tuscany.sca.binding.notification.util.IOUtils;
+import org.apache.tuscany.sca.binding.notification.util.NotificationServlet;
+import org.apache.tuscany.sca.binding.notification.util.URIUtil;
+import org.apache.tuscany.sca.binding.notification.util.IOUtils.IOUtilsException;
+import org.apache.tuscany.sca.binding.notification.util.IOUtils.Writeable;
+import org.apache.tuscany.sca.binding.notification.util.NotificationServlet.NotificationServletStreamHandler;
+import org.apache.tuscany.sca.core.invocation.MessageImpl;
+import org.apache.tuscany.sca.http.ServletHost;
+import org.apache.tuscany.sca.interfacedef.Interface;
+import org.apache.tuscany.sca.interfacedef.InterfaceContract;
+import org.apache.tuscany.sca.interfacedef.Operation;
+import org.apache.tuscany.sca.invocation.InvocationChain;
+import org.apache.tuscany.sca.invocation.Message;
+import org.apache.tuscany.sca.provider.ServiceBindingProvider;
+import org.apache.tuscany.sca.runtime.RuntimeComponent;
+import org.apache.tuscany.sca.runtime.RuntimeComponentService;
+import org.apache.tuscany.sca.runtime.RuntimeWire;
+
+/**
+ * The runtime representaion of the local service binding
+ *
+ * @version $Rev$ $Date$
+ */
+public class NotificationServiceBindingProvider
+ implements ServiceBindingProvider, NotificationServletStreamHandler {
+
+ private RuntimeWire wire;
+ private NotificationBinding notificationBinding;
+ private RuntimeComponentService service;
+ private ServletHost servletHost;
+ private NotificationTypeManager ntm;
+ private EncodingRegistry encodingRegistry;
+ private URI notificationType;
+ private URL myUrl;
+ private URL remoteNtmUrl;
+ private static final String consumerPathBase = "/consumer";
+ private boolean started;
+ private NotificationBrokerManager brokerManager;
+ private String brokerID;
+
+ public NotificationServiceBindingProvider(NotificationBinding notificationBinding,
+ RuntimeComponent component,
+ RuntimeComponentService service,
+ ServletHost servletHost,
+ NotificationTypeManager ntm,
+ EncodingRegistry encodingRegistry,
+ String httpUrl,
+ NotificationBrokerManager brokerManager) {
+ this.notificationBinding = notificationBinding;
+ this.service = service;
+ this.servletHost = servletHost;
+ this.ntm = ntm;
+ this.encodingRegistry = encodingRegistry;
+ this.notificationType = notificationBinding.getNotificationType();
+ String ntmAddress = notificationBinding.getNtmAddress();
+ String notificationTypePath = URIUtil.getPath(notificationType);
+ try {
+ this.myUrl = new URL(httpUrl + consumerPathBase + notificationTypePath);
+ remoteNtmUrl = null;
+ if (ntmAddress != null && notificationType != null) {
+ remoteNtmUrl = new URL(ntmAddress + notificationTypePath);
+ }
+ }
+ catch(Exception e) {
+ throw new RuntimeException(e);
+ }
+ this.started = false;
+ this.brokerManager = brokerManager;
+ this.brokerID = null;
+
+ URI uri = URI.create(component.getURI() + "/" + notificationBinding.getName());
+ notificationBinding.setURI(uri.toString());
+ Interface interfaze = service.getInterfaceContract().getInterface();
+ interfaze.setDefaultDataBinding(OMElement.class.getName());
+ for (Operation operation : interfaze.getOperations()) {
+ operation.setNonBlocking(false);
+ }
+ }
+
+ public NotificationBinding getBinding() {
+ return notificationBinding;
+ }
+
+ public boolean isStarted() {
+ return started;
+ }
+
+ public URL getURL() {
+ return myUrl;
+ }
+
+ public InterfaceContract getBindingInterfaceContract() {
+ return service.getInterfaceContract();
+ }
+
+ public void start() {
+ if (started) {
+ return;
+ }
+
+ RuntimeComponentService componentService = (RuntimeComponentService) service;
+ wire = componentService.getRuntimeWire(notificationBinding);
+
+ brokerManager.serviceProviderStarted(notificationType, this, remoteNtmUrl);
+ started = true;
+ }
+
+ public void stop() {
+ }
+
+ public void deployConsumer() {
+ WriteableSubscribe ws = new WriteableSubscribe(myUrl, null);
+ List<URL> producerList = new ArrayList<URL>();
+ String sequenceType = ntm.newConsumer(notificationType, myUrl, remoteNtmUrl, producerList);
+ if (Constants.EndProducers.equals(sequenceType)) {
+ for (URL producerUrl : producerList) {
+ subscribeWithProducer(producerUrl, null, ws);
+ }
+ }
+ else if (Constants.BrokerProducers.equals(sequenceType)) {
+ // Pick a broker producer, for now the first one
+ URL producerUrl = producerList.get(0);
+ subscribeWithProducer(producerUrl, null, ws);
+ }
+
+ servletHost.addServletMapping(myUrl.toString(), new NotificationServlet(this));
+ }
+
+ protected void subscribeWithProducer(URL producerUrl, String brokerID, WriteableSubscribe ws) {
+ if (ws == null) {
+ ws = new WriteableSubscribe(myUrl, brokerID);
+ }
+ try {
+ IOUtils.sendHttpRequest(producerUrl, Constants.SUBSCRIBE_OP, ws, null);
+ } catch(Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void deployBroker(String brokerID, EndpointReference brokerProducerEPR, List<EndpointReference> producerList) {
+ if (brokerProducerEPR != null) {
+ subscribeWithProducer(brokerProducerEPR.getEndpointAddress().getAddress(), brokerID, null);
+ }
+ this.brokerID = brokerID;
+ if (producerList != null && !producerList.isEmpty()) {
+ WriteableConnectionOverride wco = new WriteableConnectionOverride(myUrl, brokerID);
+ for (EndpointReference producerEPR : producerList) {
+ try {
+ IOUtils.sendHttpRequest(producerEPR.getEndpointAddress().getAddress(), Constants.CONNECTION_OVERRIDE_OP, wco, null);
+ } catch(Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ servletHost.addServletMapping(myUrl.toString(), new NotificationServlet(this));
+ }
+
+ public void replaceBrokerConnection(EndpointReference chosenBrokerProducerEpr) {
+ if (brokerID == null) {
+ throw new RuntimeException("Missing broker id");
+ }
+ URL producerUrl = chosenBrokerProducerEpr.getEndpointAddress().getAddress();
+ subscribeWithProducer(producerUrl, brokerID, null);
+ }
+
+ public void handle(Map<String, String> headers, ServletInputStream istream, int contentLength, ServletOutputStream ostream) {
+ String opHeader = headers.get(IOUtils.Notification_Operation);
+ String incomingBrokerID = headers.get(Constants.Broker_ID);
+ if (opHeader == null) {
+ throw new RuntimeException("Missing operation header");
+ }
+ if (wire == null) {
+ throw new RuntimeException("Missing wire");
+ }
+ InvocationChain chain = null;
+ for (InvocationChain ch : wire.getInvocationChains()) {
+ // We may want to use more than just the op name
+ if(ch.getTargetOperation().getName().equals(opHeader)) {
+ chain = ch;
+ break;
+ }
+ }
+ if (chain == null) {
+ throw new RuntimeException("Can't find invocation chain match for [" + opHeader + "]");
+ }
+ byte[] payload = null;
+ try {
+ payload = IOUtils.readFully(istream, contentLength);
+ }
+ catch(IOException e) {
+ throw new RuntimeException(e);
+ }
+ Object[] args = getArgsFromByteArray(payload, incomingBrokerID);
+
+ invoke(chain, args);
+
+ // Doing nothing to ostream is equivalent to returning null
+ }
+
+ private Object[] getArgsFromByteArray(byte[] payload, String incomingBrokerID) {
+ try {
+ StAXOMBuilder builder = new StAXOMBuilder(new ByteArrayInputStream(payload));
+ OMElement element = builder.getDocumentElement();
+ return new Object[] { element, payload, incomingBrokerID };
+ }
+ catch(Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected void invoke(InvocationChain chain, Object[] args) {
+ Message msg = new MessageImpl();
+ msg.setBody(args);
+ chain.getHeadInvoker().invoke(msg);
+ }
+
+ class WriteableSubscribe implements Writeable {
+
+ private Subscribe sub;
+
+ public WriteableSubscribe(URL url, String brokerID) {
+ EndpointAddress epa = new EndpointAddress();
+ epa.setAddress(url);
+ EndpointReference epr = new EndpointReference();
+ epr.setEndpointAddress(epa);
+ if (brokerID != null) {
+ BrokerID cbi = new BrokerID();
+ cbi.setID(brokerID);
+ ReferenceProperties crp = new ReferenceProperties();
+ crp.addProperty(cbi);
+ epr.setReferenceProperties(crp);
+ }
+ ConsumerReference cr = new ConsumerReference();
+ cr.setReference(epr);
+ sub = new Subscribe();
+ sub.setConsumerReference(cr);
+ }
+
+ public void write(OutputStream os) throws IOUtilsException {
+ EncodingUtils.encodeToStream(encodingRegistry, sub, os);
+ }
+ }
+
+ class WriteableConnectionOverride implements Writeable {
+
+ private ConnectionOverride connectionOverride;
+
+ public WriteableConnectionOverride(URL brokerConsumerUrl, String brokerID) {
+ EndpointAddress epa = new EndpointAddress();
+ epa.setAddress(brokerConsumerUrl);
+ EndpointReference brokerConsumerEPR = new EndpointReference();
+ brokerConsumerEPR.setEndpointAddress(epa);
+ BrokerID cbi = new BrokerID();
+ cbi.setID(brokerID);
+ ReferenceProperties crp = new ReferenceProperties();
+ crp.addProperty(cbi);
+ brokerConsumerEPR.setReferenceProperties(crp);
+ BrokerConsumerReference brokerConsumerReference = new BrokerConsumerReference();
+ brokerConsumerReference.setReference(brokerConsumerEPR);
+ connectionOverride = new ConnectionOverride();
+ connectionOverride.setBrokerConsumerReference(brokerConsumerReference);
+ }
+
+ public void write(OutputStream os) throws IOUtilsException {
+ EncodingUtils.encodeToStream(encodingRegistry, connectionOverride, os);
+ }
+ }
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationTypeManager.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationTypeManager.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationTypeManager.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationTypeManager.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,43 @@
+/*
+ * 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.sca.binding.notification;
+
+import java.net.URI;
+import java.net.URL;
+import java.util.List;
+
+import org.apache.tuscany.sca.binding.notification.encoding.EndpointReference;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public interface NotificationTypeManager {
+
+ String newConsumer(URI notificationType, URL consumerUrl, URL remoteNtmUrl, List<URL> producerList);
+ String newProducer(URI notificationType, URL producerUrl, URL remoteNtmUrl, List<URL> consumerList);
+ boolean newBroker(URI notificationType,
+ URL consumerUrl,
+ URL producerUrl,
+ String brokerID,
+ URL remoteNtmUrl,
+ List<EndpointReference> consumerList,
+ List<EndpointReference> producerList);
+ void newBrokerAck(URL remoteNtmUrl);
+ void removeBroker(EndpointReference brokerConsumerEpr, List<EndpointReference> neighborBrokerConsumerEprs, URL remoteNtmUrl);
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationTypeManagerImpl.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationTypeManagerImpl.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationTypeManagerImpl.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationTypeManagerImpl.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,692 @@
+/*
+ * 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.sca.binding.notification;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.URI;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.servlet.ServletInputStream;
+import javax.servlet.ServletOutputStream;
+
+import org.apache.tuscany.sca.binding.notification.encoding.Broker;
+import org.apache.tuscany.sca.binding.notification.encoding.BrokerConsumerReference;
+import org.apache.tuscany.sca.binding.notification.encoding.BrokerID;
+import org.apache.tuscany.sca.binding.notification.encoding.BrokerProducerReference;
+import org.apache.tuscany.sca.binding.notification.encoding.Brokers;
+import org.apache.tuscany.sca.binding.notification.encoding.Constants;
+import org.apache.tuscany.sca.binding.notification.encoding.EncodingException;
+import org.apache.tuscany.sca.binding.notification.encoding.EncodingObject;
+import org.apache.tuscany.sca.binding.notification.encoding.EncodingRegistry;
+import org.apache.tuscany.sca.binding.notification.encoding.EncodingUtils;
+import org.apache.tuscany.sca.binding.notification.encoding.EndConsumers;
+import org.apache.tuscany.sca.binding.notification.encoding.EndProducers;
+import org.apache.tuscany.sca.binding.notification.encoding.EndpointAddress;
+import org.apache.tuscany.sca.binding.notification.encoding.EndpointReference;
+import org.apache.tuscany.sca.binding.notification.encoding.EndpointReferenceWrapper;
+import org.apache.tuscany.sca.binding.notification.encoding.NeighborBrokerConsumers;
+import org.apache.tuscany.sca.binding.notification.encoding.Neighbors;
+import org.apache.tuscany.sca.binding.notification.encoding.NewBroker;
+import org.apache.tuscany.sca.binding.notification.encoding.NewBrokerAck;
+import org.apache.tuscany.sca.binding.notification.encoding.NewBrokerResponse;
+import org.apache.tuscany.sca.binding.notification.encoding.NewConsumer;
+import org.apache.tuscany.sca.binding.notification.encoding.NewConsumerResponse;
+import org.apache.tuscany.sca.binding.notification.encoding.NewProducer;
+import org.apache.tuscany.sca.binding.notification.encoding.NewProducerResponse;
+import org.apache.tuscany.sca.binding.notification.encoding.RemoveBroker;
+import org.apache.tuscany.sca.binding.notification.encoding.RemovedBroker;
+import org.apache.tuscany.sca.binding.notification.encoding.ReplaceBrokerConnection;
+import org.apache.tuscany.sca.binding.notification.util.IOUtils;
+import org.apache.tuscany.sca.binding.notification.util.NotificationServlet;
+import org.apache.tuscany.sca.binding.notification.util.URIUtil;
+import org.apache.tuscany.sca.binding.notification.util.IOUtils.IOUtilsException;
+import org.apache.tuscany.sca.binding.notification.util.IOUtils.ReadableContinuation;
+import org.apache.tuscany.sca.binding.notification.util.IOUtils.Writeable;
+import org.apache.tuscany.sca.binding.notification.util.NotificationServlet.NotificationServletStreamHandler;
+import org.apache.tuscany.sca.http.ServletHost;
+
+/**
+ * A notification type manager serves as a registry of producers and consumers, or brokers, for
+ * any notification type. This class implements an interface that allows a reference provider
+ * (a producer), a service provider (a consumer), or both (a broker, via the provider factory),
+ * to access locally the ntm for its notification type, regardless of whether the ntm resides
+ * locally or remotely.
+ * At a given host there is only one reference provider and/or one service provider for any given
+ * notification type. So, if the ntm for a notification type resides locally, then it is invoked
+ * exclusively by either a reference provider (newProducer), a service provider (newConsumer), or
+ * a provider factory (newBroker). And since these invocations occur when the providers are being
+ * created then all three of consumerLists, producerLists and brokerLists must be null when these
+ * invocations occur.
+ *
+ * @version $Rev$ $Date$
+ */
+public class NotificationTypeManagerImpl implements NotificationTypeManager {
+
+ private static final String ntmPathBase = "/ntm";
+
+ private ServletHost servletHost;
+ private EncodingRegistry encodingRegistry;
+ private Map<URI, NotificationTypeManagerHandler> ntmHandlers;
+
+ public NotificationTypeManagerImpl() {
+ }
+
+ public void setServletHost(ServletHost servletHost) {
+ this.servletHost = servletHost;
+ }
+
+ public void setEncodingRegistry(EncodingRegistry encodingRegistry) {
+ this.encodingRegistry = encodingRegistry;
+ }
+
+ public void init() {
+ ntmHandlers = new HashMap<URI, NotificationTypeManagerHandler>();
+ }
+
+ public String newConsumer(URI notificationType, URL consumerUrl, URL remoteNtmUrl, List<URL> producerListResult) {
+ if (ntmUrlIsRemote(consumerUrl, remoteNtmUrl)) {
+ try {
+ WriteableEPW wEPW = new WriteableEPW(new NewConsumer(), consumerUrl);
+ InputStreamDecoder isd = new InputStreamDecoder();
+ NewConsumerResponse ncr =
+ (NewConsumerResponse)IOUtils.sendHttpRequest(remoteNtmUrl, Constants.NEW_CONSUMER_OP, wEPW, isd);
+ String sequenceType = ncr.getSequenceType();
+ if (Constants.EndProducers.equals(sequenceType) || Constants.BrokerProducers.equals(sequenceType)) {
+ for (EndpointReference epr : ncr.getReferenceSequence()) {
+ producerListResult.add(epr.getEndpointAddress().getAddress());
+ }
+ }
+ return sequenceType;
+ } catch(Exception e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ }
+ else {
+ NotificationTypeManagerHandler ntmHandler = ntmHandlers.get(notificationType);
+ if (ntmHandler != null) {
+ throw new RuntimeException("Trying to deploy local consumer with existing local producer, consumer or broker");
+ }
+
+ createNtmHandler(consumerUrl.getAuthority(), notificationType, consumerUrl, null, null);
+
+ return Constants.NoProducers;
+ }
+ }
+
+ private void createNtmHandler(String ntmUriAuthority, URI notificationType, URL consumerUrl, URL producerUrl, BrokerStruct broker) {
+ String ntmUri = "http://" + ntmUriAuthority + ntmPathBase + URIUtil.getPath(notificationType);
+ NotificationTypeManagerHandler ntmh = new NotificationTypeManagerHandler(notificationType, consumerUrl, producerUrl, broker);
+ ntmHandlers.put(notificationType, ntmh);
+ servletHost.addServletMapping(ntmUri, new NotificationServlet(ntmh));
+ }
+
+ public String newProducer(URI notificationType, URL producerUrl, URL remoteNtmUrl, List<URL> consumerListResult) {
+ if (ntmUrlIsRemote(producerUrl, remoteNtmUrl)) {
+ try {
+ WriteableEPW wEPW = new WriteableEPW(new NewProducer(), producerUrl);
+ InputStreamDecoder isd = new InputStreamDecoder();
+ NewProducerResponse npr =
+ (NewProducerResponse)IOUtils.sendHttpRequest(remoteNtmUrl, Constants.NEW_PRODUCER_OP, wEPW, isd);
+ String sequenceType = npr.getSequenceType();
+ if (Constants.EndConsumers.equals(sequenceType) || Constants.BrokerConsumers.equals(sequenceType)) {
+ for (EndpointReference epr : npr.getReferenceSequence()) {
+ consumerListResult.add(epr.getEndpointAddress().getAddress());
+ }
+ }
+ return sequenceType;
+ } catch(Exception e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ }
+ else {
+ NotificationTypeManagerHandler ntmHandler = ntmHandlers.get(notificationType);
+ if (ntmHandler != null) {
+ throw new RuntimeException("Trying to deploy local producer with existing local producer, consumer or broker");
+ }
+
+ createNtmHandler(producerUrl.getAuthority(), notificationType, null, producerUrl, null);
+
+ return Constants.NoConsumers;
+ }
+ }
+
+ public boolean newBroker(URI notificationType,
+ URL consumerUrl,
+ URL producerUrl,
+ String brokerID,
+ URL remoteNtmUrl,
+ List<EndpointReference> consumerListResult,
+ List<EndpointReference> producerListResult) {
+ String ntmUriAuthority = producerUrl.getAuthority();
+ if (!ntmUriAuthority.equals(consumerUrl.getAuthority())) {
+ throw new RuntimeException("Producer url and consumer url do not match");
+ }
+ if (ntmUrlIsRemote(producerUrl, remoteNtmUrl)) {
+ try {
+ WriteableNewBroker wnb = new WriteableNewBroker(consumerUrl, producerUrl, brokerID);
+ InputStreamDecoder isd = new InputStreamDecoder();
+ NewBrokerResponse nbr =
+ (NewBrokerResponse)IOUtils.sendHttpRequest(remoteNtmUrl, Constants.NEW_BROKER_OP, wnb, isd);
+ if (nbr.isFirstBroker()) {
+ if (nbr.getEndConsumers().getSequenceType().equals(Constants.EndConsumers)) {
+ for (EndpointReference epr : nbr.getEndConsumers().getReferenceSequence()) {
+ consumerListResult.add(epr);
+ }
+ }
+ if (nbr.getEndProducers().getSequenceType().equals(Constants.EndProducers)) {
+ for (EndpointReference epr : nbr.getEndProducers().getReferenceSequence()) {
+ producerListResult.add(epr);
+ }
+ }
+ }
+ else {
+ for (Broker broker : nbr.getBrokers().getBrokerSequence()) {
+ consumerListResult.add(broker.getBrokerConsumerReference().getReference());
+ producerListResult.add(broker.getBrokerProducerReference().getReference());
+ }
+ }
+ return nbr.isFirstBroker();
+ } catch(Exception e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ }
+ else {
+ NotificationTypeManagerHandler ntmHandler = ntmHandlers.get(notificationType);
+ if (ntmHandler != null) {
+ throw new RuntimeException("Trying to deploy local broker with existing local producer, consumer or broker");
+ }
+
+ BrokerStruct broker = new BrokerStruct(consumerUrl, producerUrl, brokerID);
+ createNtmHandler(ntmUriAuthority, notificationType, null, null, broker);
+
+ return true;
+ }
+ }
+
+ private boolean ntmUrlIsRemote(URL localUrl, URL ntmUrl) {
+ if (ntmUrl == null) {
+ return false;
+ }
+ if (localUrl.getPort() != ntmUrl.getPort()) {
+ return true;
+ }
+ String remoteNtmUrlAuthority = ntmUrl.getAuthority();
+ if (remoteNtmUrlAuthority.indexOf("localhost") >= 0) {
+ return false;
+ }
+ return !localUrl.getAuthority().equals(remoteNtmUrlAuthority);
+ }
+
+ public void newBrokerAck(URL remoteNtmUrl) {
+ try {
+ IOUtils.sendHttpRequest(remoteNtmUrl, Constants.NEW_BROKER_ACK_OP, new WriteableNewBrokerAck(), null);
+ } catch(Exception e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void removeBroker(EndpointReference brokerConsumerEpr, List<EndpointReference> neighborBrokerConsumerEprs, URL remoteNtmUrl) {
+ WriteableRemoveBroker wrb = new WriteableRemoveBroker(brokerConsumerEpr, neighborBrokerConsumerEprs);
+
+ try {
+ IOUtils.sendHttpRequest(remoteNtmUrl, Constants.REMOVE_BROKER_OP, wrb, null);
+ } catch(Exception e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ }
+
+ private class NotificationTypeManagerHandler implements NotificationServletStreamHandler {
+
+ private URI notificationType;
+ List<URL> consumerList;
+ List<URL> producerList;
+ List<BrokerStruct> brokerList;
+ private NotificationTypeLock notificationTypeLock;
+ private BrokerStruct pendingBroker;
+
+ public NotificationTypeManagerHandler(URI notificationType) {
+ this.notificationType = notificationType;
+ this.notificationTypeLock = new NotificationTypeLock();
+ this.pendingBroker = null;
+ }
+
+ public NotificationTypeManagerHandler(URI notificationType, URL consumerUrl, URL producerUrl, BrokerStruct broker) {
+ this(notificationType);
+ if (consumerUrl != null) {
+ addConsumer(consumerUrl);
+ }
+ else if (producerUrl != null) {
+ addProducer(producerUrl);
+ }
+ else if (broker != null) {
+ addBroker(broker);
+ }
+ }
+
+ private void addConsumer(URL consumerUrl) {
+ if (consumerList == null) {
+ consumerList = new ArrayList<URL>();
+ }
+ consumerList.add(consumerUrl);
+ }
+
+ private void addProducer(URL producerUrl) {
+ if (producerList == null) {
+ producerList = new ArrayList<URL>();
+ }
+ producerList.add(producerUrl);
+ }
+
+ private void addBroker(BrokerStruct broker) {
+ if (brokerList == null) {
+ brokerList = new ArrayList<BrokerStruct>();
+ }
+ brokerList.add(broker);
+ }
+
+ public void handle(Map<String, String> headers, ServletInputStream istream, int contentLength, ServletOutputStream ostream) {
+ String opHeader = headers.get(IOUtils.Notification_Operation);
+ EncodingObject eo = null;
+ try {
+ eo = EncodingUtils.decodeFromStream(encodingRegistry, istream);
+ }
+ catch(EncodingException e) {
+ throw new RuntimeException(e);
+ }
+
+ if (Constants.NEW_CONSUMER_OP.equals(opHeader)) {
+ handleNewConsumer((NewConsumer)eo, ostream);
+ }
+ else if(Constants.NEW_PRODUCER_OP.equals(opHeader)) {
+ handleNewProducer((NewProducer)eo, ostream);
+ }
+ else if(Constants.NEW_BROKER_OP.equals(opHeader)) {
+ handleNewBroker((NewBroker)eo, ostream);
+ }
+ else if (Constants.NEW_BROKER_ACK_OP.equals(opHeader)) {
+ handleNewBrokerAck();
+ }
+ else if (Constants.REMOVE_BROKER_OP.equals(opHeader)) {
+ handleRemoveBroker((RemoveBroker)eo);
+ }
+ }
+
+ private void handleNewConsumer(NewConsumer nc, ServletOutputStream ostream) {
+ synchronized(notificationTypeLock) {
+ if (notificationTypeLock.isLocked) {
+ try { notificationTypeLock.wait(); } catch(InterruptedException e) {}
+ }
+ URL consumerUrl = nc.getReference().getEndpointAddress().getAddress();
+ if (brokerList == null) {
+ addConsumer(consumerUrl);
+ }
+
+ NewConsumerResponse ncr = new NewConsumerResponse();
+ if (producerList != null) {
+ ncr.setSequenceType(Constants.EndProducers);
+ for (URL producerUrl : producerList) {
+ ncr.addReferenceToSequence(EncodingUtils.createEndpointReference(producerUrl, null));
+ }
+ }
+ else if(brokerList != null) {
+ ncr.setSequenceType(Constants.BrokerProducers);
+ for (BrokerStruct broker : brokerList) {
+ ncr.addReferenceToSequence(EncodingUtils.createEndpointReference(broker.producerUrl, null));
+ }
+ }
+ else {
+ ncr.setSequenceType(Constants.NoProducers);
+ }
+ try {
+ EncodingUtils.encodeToStream(encodingRegistry, ncr, ostream);
+ }
+ catch(IOUtilsException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ private void handleNewProducer(NewProducer np, ServletOutputStream ostream) {
+ synchronized(notificationTypeLock) {
+ if (notificationTypeLock.isLocked) {
+ try { notificationTypeLock.wait(); } catch(InterruptedException e) {}
+ }
+ URL producerUrl = np.getReference().getEndpointAddress().getAddress();
+ if (brokerList == null) {
+ addProducer(producerUrl);
+ }
+
+ NewProducerResponse npr = new NewProducerResponse();
+ if (consumerList != null) {
+ npr.setSequenceType(Constants.EndConsumers);
+ for (URL consumerUrl : consumerList) {
+ npr.addReferenceToSequence(EncodingUtils.createEndpointReference(consumerUrl, null));
+ }
+ }
+ else if(brokerList != null) {
+ npr.setSequenceType(Constants.BrokerConsumers);
+ for (BrokerStruct broker : brokerList) {
+ npr.addReferenceToSequence(EncodingUtils.createEndpointReference(broker.consumerUrl, null));
+ }
+ }
+ else {
+ npr.setSequenceType(Constants.NoConsumers);
+ }
+ try {
+ EncodingUtils.encodeToStream(encodingRegistry, npr, ostream);
+ }
+ catch(IOUtilsException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ private void handleNewBroker(NewBroker nb, ServletOutputStream ostream) {
+ synchronized(notificationTypeLock) {
+ if (notificationTypeLock.isLocked) {
+ try { notificationTypeLock.wait(); } catch(InterruptedException e) {}
+ }
+ NewBrokerResponse nbr = new NewBrokerResponse();
+ if (consumerList != null || producerList != null || brokerList == null) {
+ nbr.setFirstBroker(true);
+ EndConsumers endConsumers = new EndConsumers();
+ if (consumerList != null) {
+ endConsumers.setSequenceType(Constants.EndConsumers);
+ for (URL consumerUrl : consumerList) {
+ endConsumers.addReferenceToSequence(EncodingUtils.createEndpointReference(consumerUrl, null));
+ }
+ }
+ else {
+ endConsumers.setSequenceType(Constants.NoConsumers);
+ }
+ nbr.setEndConsumers(endConsumers);
+ EndProducers endProducers = new EndProducers();
+ if (producerList != null) {
+ endProducers.setSequenceType(Constants.EndProducers);
+ for (URL producerUrl : producerList) {
+ endProducers.addReferenceToSequence(EncodingUtils.createEndpointReference(producerUrl, null));
+ }
+ }
+ else {
+ endProducers.setSequenceType(Constants.NoProducers);
+ }
+ nbr.setEndProducers(endProducers);
+ }
+ else {
+ nbr.setFirstBroker(false);
+ Brokers brokers = new Brokers();
+ for (BrokerStruct brokerStruct : brokerList) {
+ Broker brokerElt = new Broker();
+ BrokerConsumerReference bcr = new BrokerConsumerReference();
+ bcr.setReference(EncodingUtils.createEndpointReference(brokerStruct.consumerUrl, brokerStruct.brokerID));
+ brokerElt.setBrokerConsumerReference(bcr);
+
+ BrokerProducerReference bpr = new BrokerProducerReference();
+ bpr.setReference(EncodingUtils.createEndpointReference(brokerStruct.producerUrl, brokerStruct.brokerID));
+ brokerElt.setBrokerProducerReference(bpr);
+ brokers.addBrokerToSequence(brokerElt);
+ }
+ nbr.setBrokers(brokers);
+ }
+ EndpointReference consumerEPR = nb.getBrokerConsumerReference().getReference();
+ URL consumerUrl = consumerEPR.getEndpointAddress().getAddress();
+ BrokerID consumerBrokerID = consumerEPR.getReferenceProperties().getProperty(BrokerID.class);
+ EndpointReference producerEPR = nb.getBrokerProducerReference().getReference();
+ URL producerUrl = producerEPR.getEndpointAddress().getAddress();
+ BrokerID producerBrokerID = producerEPR.getReferenceProperties().getProperty(BrokerID.class);
+ if (consumerBrokerID == null ||
+ producerBrokerID == null ||
+ !consumerBrokerID.getID().equals(producerBrokerID.getID())) {
+ throw new RuntimeException("Producer and consumer broker ids do not match");
+ }
+ // only add broker if consumerList == null && producerList == null
+ // otherwise, make it a pending broker and wait for ack
+ // TODO block for a configurable amount of time
+ BrokerStruct broker = new BrokerStruct(consumerUrl, producerUrl, consumerBrokerID.getID());
+ if (consumerList == null && producerList == null) {
+ addBroker(broker);
+ }
+ else {
+ pendingBroker = broker;
+ notificationTypeLock.isLocked = true;
+ }
+ try {
+ EncodingUtils.encodeToStream(encodingRegistry, nbr, ostream);
+ }
+ catch(IOUtilsException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ private void handleNewBrokerAck() {
+ synchronized(notificationTypeLock) {
+ if (!notificationTypeLock.isLocked) {
+ notificationTypeLock.notifyAll();
+ throw new RuntimeException("Notification type should be locked");
+ }
+ if (brokerList != null) {
+ notificationTypeLock.isLocked = false;
+ notificationTypeLock.notifyAll();
+ throw new RuntimeException("Can't add pending broker to non-empty broker list");
+ }
+ if (pendingBroker == null) {
+ notificationTypeLock.isLocked = false;
+ notificationTypeLock.notifyAll();
+ throw new RuntimeException("Missing pending broker");
+ }
+ addBroker(pendingBroker);
+ consumerList = null;
+ producerList = null;
+ pendingBroker = null;
+ notificationTypeLock.isLocked = false;
+ notificationTypeLock.notifyAll();
+ }
+ }
+
+ private void handleRemoveBroker(RemoveBroker rb) {
+ synchronized(notificationTypeLock) {
+ if (notificationTypeLock.isLocked) {
+ try { notificationTypeLock.wait(); } catch(InterruptedException e) {}
+ }
+
+ if (brokerList == null) {
+ throw new RuntimeException("No broker to remove for [" + notificationType + "]");
+ }
+
+ NeighborBrokerConsumers nbcs = rb.getNeighborBrokerConsumers();
+ EndpointReference rbEpr = rb.getBrokerConsumerReference().getReference();
+ if (nbcs != null && nbcs.getReferenceSequence() != null) {
+ List<Broker> neighborBrokers = new ArrayList<Broker>();
+ for (EndpointReference neighborBrokerConsumerEpr : nbcs.getReferenceSequence()) {
+ BrokerStruct neighborBrokerStruct = null;
+ URL neighborBrokerConsumerEprUrl = neighborBrokerConsumerEpr.getEndpointAddress().getAddress();
+ for (BrokerStruct brokerStruct : brokerList) {
+ if (brokerStruct.consumerUrl.equals(neighborBrokerConsumerEprUrl)) {
+ neighborBrokerStruct = brokerStruct;
+ break;
+ }
+ }
+ if (neighborBrokerStruct == null) {
+ throw new RuntimeException("Can't find neighbor broker for consumer EPR [" +
+ neighborBrokerConsumerEprUrl + "]");
+ }
+ BrokerConsumerReference bcr = new BrokerConsumerReference();
+ bcr.setReference(EncodingUtils.createEndpointReference(neighborBrokerStruct.consumerUrl, neighborBrokerStruct.brokerID));
+ BrokerProducerReference bpr = new BrokerProducerReference();
+ bpr.setReference(EncodingUtils.createEndpointReference(neighborBrokerStruct.producerUrl, neighborBrokerStruct.brokerID));
+ Broker neighborBroker = new Broker();
+ neighborBroker.setBrokerConsumerReference(bcr);
+ neighborBroker.setBrokerProducerReference(bpr);
+ neighborBrokers.add(neighborBroker);
+ }
+ int lastIndex = neighborBrokers.size() - 1;
+ for (int index = lastIndex; index >= 0; index--) {
+ List<Broker> writeableNeighborBrokers = ((index > 0) ? neighborBrokers.subList(0, index) : null);
+ WriteableReplaceBrokerConnection wrbc = new WriteableReplaceBrokerConnection(rbEpr, writeableNeighborBrokers);
+ URL targetUrl =
+ neighborBrokers.get(index).getBrokerProducerReference().getReference().getEndpointAddress().getAddress();
+ try {
+ IOUtils.sendHttpRequest(targetUrl, Constants.REPLACE_BROKER_CONNECTION_OP, wrbc, null);
+ } catch(Exception e) {
+ e.printStackTrace();
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ BrokerStruct removedBrokerStruct = null;
+ URL rbEprUrl = rbEpr.getEndpointAddress().getAddress();
+ for (BrokerStruct brokerSruct : brokerList) {
+ if (brokerSruct.consumerUrl.equals(rbEprUrl)) {
+ removedBrokerStruct = brokerSruct;
+ break;
+ }
+ }
+ if (removedBrokerStruct == null) {
+ throw new RuntimeException("Can't find broker to remove for EPR [" + rbEprUrl + "]");
+ }
+ if(!brokerList.remove(removedBrokerStruct)) {
+ throw new RuntimeException("Broker was not removed");
+ }
+ }
+ }
+ }
+
+ class NotificationTypeLock {
+ public boolean isLocked;
+ }
+
+ class WriteableEPW implements Writeable {
+ private EndpointReferenceWrapper epw;
+
+ public WriteableEPW(EndpointReferenceWrapper epw, URL url) {
+ EndpointAddress epa = new EndpointAddress();
+ epa.setAddress(url);
+ EndpointReference epr = new EndpointReference();
+ epr.setEndpointAddress(epa);
+ epw.setReference(epr);
+ this.epw = epw;
+ }
+
+ public void write(OutputStream os) throws IOUtilsException {
+ EncodingUtils.encodeToStream(encodingRegistry, epw, os);
+ }
+ }
+
+ class InputStreamDecoder implements ReadableContinuation {
+
+ public Object read(InputStream istream) throws IOUtilsException {
+ try {
+ return EncodingUtils.decodeFromStream(encodingRegistry, istream);
+ }
+ catch(EncodingException e) {
+ throw new IOUtilsException(e);
+ }
+ }
+ }
+
+ class BrokerStruct {
+ public URL consumerUrl;
+ public URL producerUrl;
+ public String brokerID;
+
+ public BrokerStruct(URL consumerUrl, URL producerUrl, String brokerID) {
+ this.consumerUrl = consumerUrl;
+ this.producerUrl = producerUrl;
+ this.brokerID = brokerID;
+ }
+ }
+
+ class WriteableNewBroker implements Writeable {
+ private NewBroker newBroker;
+
+ public WriteableNewBroker(URL consumerUrl, URL producerUrl, String brokerID) {
+ newBroker = new NewBroker();
+ BrokerConsumerReference bcr = new BrokerConsumerReference();
+ bcr.setReference(EncodingUtils.createEndpointReference(consumerUrl, brokerID));
+ newBroker.setBrokerConsumerReference(bcr);
+
+ BrokerProducerReference bpr = new BrokerProducerReference();
+ bpr.setReference(EncodingUtils.createEndpointReference(producerUrl, brokerID));
+ newBroker.setBrokerProducerReference(bpr);
+ }
+
+ public void write(OutputStream os) throws IOUtilsException {
+ EncodingUtils.encodeToStream(encodingRegistry, newBroker, os);
+ }
+ }
+
+ class WriteableNewBrokerAck implements Writeable {
+ public void write(OutputStream os) throws IOUtilsException {
+ EncodingUtils.encodeToStream(encodingRegistry, new NewBrokerAck(), os);
+ }
+ }
+
+ class WriteableRemoveBroker implements Writeable {
+ private RemoveBroker removeBroker;
+
+ public WriteableRemoveBroker(EndpointReference brokerConsumerEpr, List<EndpointReference> neighborBrokerConsumerEprs) {
+ removeBroker = new RemoveBroker();
+ BrokerConsumerReference brokerConsumerReference = new BrokerConsumerReference();
+ brokerConsumerReference.setReference(brokerConsumerEpr);
+ removeBroker.setBrokerConsumerReference(brokerConsumerReference);
+ if (neighborBrokerConsumerEprs != null) {
+ NeighborBrokerConsumers neighborBrokerConsumers = new NeighborBrokerConsumers();
+ neighborBrokerConsumers.setReferenceSequence(neighborBrokerConsumerEprs);
+ neighborBrokerConsumers.setSequenceType(Constants.BrokerConsumers);
+ removeBroker.setNeighborBrokerConsumers(neighborBrokerConsumers);
+ }
+ }
+
+ public void write(OutputStream os) throws IOUtilsException {
+ EncodingUtils.encodeToStream(encodingRegistry, removeBroker, os);
+ }
+ }
+
+ class WriteableReplaceBrokerConnection implements Writeable {
+ private ReplaceBrokerConnection replaceBrokerConnection;
+
+ public WriteableReplaceBrokerConnection(EndpointReference removedBrokerEpr, List<Broker> brokerSequence) {
+ replaceBrokerConnection = new ReplaceBrokerConnection();
+ RemovedBroker removedBroker = new RemovedBroker();
+ removedBroker.setReference(removedBrokerEpr);
+ replaceBrokerConnection.setRemovedBroker(removedBroker);
+ if (brokerSequence != null) {
+ Neighbors neighbors = new Neighbors();
+ neighbors.setBrokerSequence(brokerSequence);
+ replaceBrokerConnection.setNeighbors(neighbors);
+ }
+ }
+
+ public void write(OutputStream os) throws IOUtilsException {
+ EncodingUtils.encodeToStream(encodingRegistry, replaceBrokerConnection, os);
+ }
+ }
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractBroker.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractBroker.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractBroker.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractBroker.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,44 @@
+/*
+ * 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.sca.binding.notification.encoding;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class AbstractBroker implements EncodingObject {
+
+ private BrokerProducerReference brokerProducerReference;
+ private BrokerConsumerReference brokerConsumerReference;
+
+ public BrokerProducerReference getBrokerProducerReference() {
+ return this.brokerProducerReference;
+ }
+
+ public void setBrokerProducerReference(BrokerProducerReference brokerProducerReference) {
+ this.brokerProducerReference = brokerProducerReference;
+ }
+
+ public BrokerConsumerReference getBrokerConsumerReference() {
+ return this.brokerConsumerReference;
+ }
+
+ public void setBrokerConsumerReference(BrokerConsumerReference brokerConsumerReference) {
+ this.brokerConsumerReference = brokerConsumerReference;
+ }
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractBrokerEnDeCoder.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractBrokerEnDeCoder.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractBrokerEnDeCoder.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractBrokerEnDeCoder.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,88 @@
+/*
+ * 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.sca.binding.notification.encoding;
+
+import static javax.xml.stream.XMLStreamConstants.END_ELEMENT;
+import static javax.xml.stream.XMLStreamConstants.START_ELEMENT;
+
+import javax.xml.namespace.QName;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.XMLStreamReader;
+import javax.xml.stream.XMLStreamWriter;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public abstract class AbstractBrokerEnDeCoder<B extends AbstractBroker> extends AbstractEnDeCoder<B> {
+
+ public AbstractBrokerEnDeCoder(EncodingRegistry registry) {
+ super(registry);
+ }
+
+ public void encode(B encodingObject, XMLStreamWriter writer) throws EncodingException {
+
+ try {
+ QName qName = getEncodingObjectQName();
+ writer.writeStartElement(Constants.NOTIFICATION_PREFIX, qName.getLocalPart(), qName.getNamespaceURI());
+ writer.writeNamespace(Constants.NOTIFICATION_PREFIX, qName.getNamespaceURI());
+ registry.encode(encodingObject.getBrokerConsumerReference(), writer);
+ registry.encode(encodingObject.getBrokerProducerReference(), writer);
+ writer.writeEndElement();
+ } catch(XMLStreamException e) {
+ throw new EncodingException(e);
+ }
+ }
+
+ public B decode(XMLStreamReader reader) throws EncodingException {
+
+ try {
+ B brokerElement = getEncodingObjectType().newInstance();
+ boolean haveBCR = false;
+ boolean haveBPR = false;
+ while (true) {
+ switch (reader.next()) {
+ case START_ELEMENT:
+ EncodingObject encodingObject = registry.decode(reader);
+ if (encodingObject instanceof BrokerConsumerReference && !haveBCR) {
+ brokerElement.setBrokerConsumerReference((BrokerConsumerReference)encodingObject);
+ haveBCR = true;
+ }
+ else if(encodingObject instanceof BrokerProducerReference && !haveBPR) {
+ brokerElement.setBrokerProducerReference((BrokerProducerReference)encodingObject);
+ haveBPR = true;
+ }
+ else {
+ throw new EncodingException("Invalid encoding object");
+ }
+ break;
+ case END_ELEMENT:
+ if (!haveBCR) {
+ throw new EncodingException("Missing broker consumer reference");
+ }
+ if (!haveBPR) {
+ throw new EncodingException("Missing broker producer reference");
+ }
+ return brokerElement;
+ }
+ }
+ } catch (Exception ex) {
+ throw new EncodingException(ex);
+ }
+ }
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractEnDeCoder.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractEnDeCoder.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractEnDeCoder.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/AbstractEnDeCoder.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,63 @@
+/*
+ * 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.sca.binding.notification.encoding;
+
+import javax.xml.namespace.QName;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public abstract class AbstractEnDeCoder<E extends EncodingObject> implements EnDeCoder<E> {
+
+ protected EncodingRegistry registry;
+
+ protected AbstractEnDeCoder(EncodingRegistry registry) {
+
+ this.registry = registry;
+ }
+
+ public void start() {
+ Class<E> encodingType = getEncodingObjectType();
+ QName encodingQName = getEncodingObjectQName();
+
+ registry.registerEnDeCoder(encodingType, encodingQName, this);
+ }
+
+ public void stop() {
+ Class<E> encodingType = getEncodingObjectType();
+ QName encodingQName = getEncodingObjectQName();
+
+ registry.unregisterEnDeCoder(encodingType, encodingQName);
+ }
+
+ /**
+ * Gets the qualified name of the XML fragment for the Encoding
+ * object.
+ *
+ * @return Qualified name of the XML fragment.
+ */
+ protected abstract QName getEncodingObjectQName();
+
+ /**
+ * Returns the type of the encoding object.
+ *
+ * @return Encoding object type.
+ */
+ protected abstract Class<E> getEncodingObjectType();
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/Broker.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/Broker.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/Broker.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/Broker.java Thu Jun 28 09:30:41 2007
@@ -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.
+ */
+package org.apache.tuscany.sca.binding.notification.encoding;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class Broker extends AbstractBroker {
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerConsumerReference.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerConsumerReference.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerConsumerReference.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerConsumerReference.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,26 @@
+/*
+ * 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.sca.binding.notification.encoding;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class BrokerConsumerReference extends EndpointReferenceWrapper {
+
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerConsumerReferenceEnDeCoder.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerConsumerReferenceEnDeCoder.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerConsumerReferenceEnDeCoder.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerConsumerReferenceEnDeCoder.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,42 @@
+/*
+ * 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.sca.binding.notification.encoding;
+
+import javax.xml.namespace.QName;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class BrokerConsumerReferenceEnDeCoder extends EndpointReferenceWrapperEnDeCoder<BrokerConsumerReference> {
+
+ // QName for the root element
+ public static final QName QNAME = new QName(Constants.NOTIFICATION_NS, Constants.BrokerConsumerReference);
+
+ public BrokerConsumerReferenceEnDeCoder(EncodingRegistry registry) {
+ super(registry);
+ }
+
+ protected QName getEncodingObjectQName() {
+ return QNAME;
+ }
+
+ protected Class<BrokerConsumerReference> getEncodingObjectType() {
+ return BrokerConsumerReference.class;
+ }
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerEnDeCoder.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerEnDeCoder.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerEnDeCoder.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerEnDeCoder.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,44 @@
+/*
+ * 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.sca.binding.notification.encoding;
+
+import javax.xml.namespace.QName;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class BrokerEnDeCoder extends AbstractBrokerEnDeCoder<Broker> {
+
+ // QName for the root element
+ public static final QName QNAME = new QName(Constants.NOTIFICATION_NS, Constants.Broker);
+
+ public BrokerEnDeCoder(EncodingRegistry registry) {
+ super(registry);
+ }
+
+ protected QName getEncodingObjectQName() {
+
+ return QNAME;
+ }
+
+ protected Class<Broker> getEncodingObjectType() {
+
+ return Broker.class;
+ }
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerID.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerID.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerID.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerID.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,41 @@
+/*
+ * 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.sca.binding.notification.encoding;
+
+import java.util.UUID;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class BrokerID implements EncodingObject {
+
+ private String id;
+
+ public String getID() {
+ return id;
+ }
+
+ public void setID(String id) {
+ this.id = id;
+ }
+
+ public static String generate() {
+ return UUID.randomUUID().toString();
+ }
+}
Added: incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerIDEnDeCoder.java
URL: http://svn.apache.org/viewvc/incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerIDEnDeCoder.java?view=auto&rev=551614
==============================================================================
--- incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerIDEnDeCoder.java (added)
+++ incubator/tuscany/java/sca/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/encoding/BrokerIDEnDeCoder.java Thu Jun 28 09:30:41 2007
@@ -0,0 +1,89 @@
+/*
+ * 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.sca.binding.notification.encoding;
+
+import static javax.xml.stream.XMLStreamConstants.END_ELEMENT;
+import static javax.xml.stream.XMLStreamConstants.START_ELEMENT;
+
+import javax.xml.namespace.QName;
+import javax.xml.stream.XMLStreamConstants;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.XMLStreamReader;
+import javax.xml.stream.XMLStreamWriter;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class BrokerIDEnDeCoder extends AbstractEnDeCoder<BrokerID> {
+
+ // QName for the root element
+ public static final QName QNAME = new QName(Constants.NOTIFICATION_NS, Constants.BrokerID);
+
+ public BrokerIDEnDeCoder(EncodingRegistry registry) {
+ super(registry);
+ }
+
+ public void encode(BrokerID encodingObject, XMLStreamWriter writer) throws EncodingException {
+
+ try {
+ writer.writeStartElement(Constants.NOTIFICATION_PREFIX, QNAME.getLocalPart(), QNAME.getNamespaceURI());
+ writer.writeNamespace(Constants.NOTIFICATION_PREFIX, QNAME.getNamespaceURI());
+ writer.writeCharacters(encodingObject.getID());
+ writer.writeEndElement();
+ } catch(XMLStreamException e) {
+ throw new EncodingException(e);
+ }
+ }
+
+ public BrokerID decode(XMLStreamReader reader) throws EncodingException {
+
+ try {
+ BrokerID brokerIDElement = new BrokerID();
+ while (true) {
+ switch (reader.next()) {
+ case START_ELEMENT:
+ break;
+ case XMLStreamConstants.CHARACTERS:
+ if (reader.hasText()) {
+ String id = reader.getText();
+ brokerIDElement.setID(id);
+ }
+ else {
+ throw new EncodingException("Broker ID missing value");
+ }
+ break;
+ case END_ELEMENT:
+ return brokerIDElement;
+ }
+ }
+ } catch (XMLStreamException ex) {
+ throw new EncodingException(ex);
+ }
+ }
+
+ protected QName getEncodingObjectQName() {
+
+ return QNAME;
+ }
+
+ protected Class<BrokerID> getEncodingObjectType() {
+
+ return BrokerID.class;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: tuscany-commits-unsubscribe@ws.apache.org
For additional commands, e-mail: tuscany-commits-help@ws.apache.org