diff options
author | lresende <lresende@13f79535-47bb-0310-9956-ffa450edef68> | 2009-11-11 23:26:33 +0000 |
---|---|---|
committer | lresende <lresende@13f79535-47bb-0310-9956-ffa450edef68> | 2009-11-11 23:26:33 +0000 |
commit | a40e527938d76ba71f211da7e327adb50384ba69 (patch) | |
tree | cb8f99f1727122b040a3f0fbb6649292b6a74302 /sca-java-1.x/tags/1.2/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationBindingProviderFactory.java | |
parent | 968721109881107520d7aefa91d7fcc0519d7739 (diff) |
Moving 1.x tags
git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@835157 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'sca-java-1.x/tags/1.2/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationBindingProviderFactory.java')
-rw-r--r-- | sca-java-1.x/tags/1.2/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationBindingProviderFactory.java | 334 |
1 files changed, 334 insertions, 0 deletions
diff --git a/sca-java-1.x/tags/1.2/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationBindingProviderFactory.java b/sca-java-1.x/tags/1.2/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationBindingProviderFactory.java new file mode 100644 index 0000000000..7592f4f824 --- /dev/null +++ b/sca-java-1.x/tags/1.2/modules/binding-notification/src/main/java/org/apache/tuscany/sca/binding/notification/NotificationBindingProviderFactory.java @@ -0,0 +1,334 @@ +/* + * 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.InetAddress; +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 org.apache.tuscany.sca.binding.notification.encoding.AbstractEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.BrokerConsumerReferenceEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.BrokerEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.BrokerID; +import org.apache.tuscany.sca.binding.notification.encoding.BrokerIDEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.BrokerProducerReferenceEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.BrokersEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.ConnectionOverrideEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.ConnectionOverrideResponseEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.ConsumerReferenceEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.EncodingRegistry; +import org.apache.tuscany.sca.binding.notification.encoding.EndConsumersEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.EndProducersEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.EndpointAddressEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.EndpointReference; +import org.apache.tuscany.sca.binding.notification.encoding.EndpointReferenceEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.NeighborBrokerConsumersEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.NeighborsEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.NewBrokerAckEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.NewBrokerEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.NewBrokerResponseEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.NewConsumerEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.NewConsumerResponseEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.NewProducerEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.NewProducerResponseEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.ReferencePropertiesEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.RemoveBrokerEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.RemovedBrokerEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.ReplaceBrokerConnectionEnDeCoder; +import org.apache.tuscany.sca.binding.notification.encoding.SubscribeEnDeCoder; +import org.apache.tuscany.sca.host.http.ServletHost; +import org.apache.tuscany.sca.provider.BindingProviderFactory; +import org.apache.tuscany.sca.provider.ReferenceBindingProvider; +import org.apache.tuscany.sca.provider.ServiceBindingProvider; +import org.apache.tuscany.sca.runtime.RuntimeComponent; +import org.apache.tuscany.sca.runtime.RuntimeComponentReference; +import org.apache.tuscany.sca.runtime.RuntimeComponentService; + +/** + * The runtime representation of the notification reference binding + * + * @version $Rev$ $Date$ + */ +public class NotificationBindingProviderFactory implements BindingProviderFactory<NotificationBinding>, + NotificationBrokerManager { + private static final String DEFAULT_PORT = "8083"; + + private ServletHost servletHost; + private NotificationTypeManagerImpl notificationTypeManager; + private EncodingRegistry encodingRegistry; + private String httpUrl; + private Map<URI, NotificationReferenceBindingProvider> referenceBindingProviders; + private Map<URI, NotificationServiceBindingProvider> serviceBindingProviders; + + private static NotificationBindingProviderFactory factoryInstance = null; + + public NotificationBindingProviderFactory(ServletHost servletHost, EncodingRegistry encodingRegistry) { + this.servletHost = servletHost; + this.encodingRegistry = encodingRegistry; + this.referenceBindingProviders = new HashMap<URI, NotificationReferenceBindingProvider>(); + this.serviceBindingProviders = new HashMap<URI, NotificationServiceBindingProvider>(); + + factoryInstance = this; + } + + public Class<NotificationBinding> getModelType() { + return NotificationBinding.class; + } + + public ReferenceBindingProvider createReferenceBindingProvider(RuntimeComponent component, + RuntimeComponentReference reference, + NotificationBinding binding) { + init(); + URI notificationType = binding.getNotificationType(); + if (!validReferenceBinding(binding)) { + throw new RuntimeException("Binding not valid"); + } + NotificationReferenceBindingProvider referenceBindingProvider = referenceBindingProviders.get(notificationType); + if (referenceBindingProvider == null) { + referenceBindingProvider = + new NotificationReferenceBindingProvider(binding, component, reference, servletHost, + notificationTypeManager, encodingRegistry, httpUrl, this); + referenceBindingProviders.put(notificationType, referenceBindingProvider); + } + return referenceBindingProvider; + } + + public ServiceBindingProvider createServiceBindingProvider(RuntimeComponent component, + RuntimeComponentService service, + NotificationBinding binding) { + init(); + URI notificationType = binding.getNotificationType(); + if (!validServiceBinding(binding)) { + throw new RuntimeException("Binding not valid"); + } + NotificationServiceBindingProvider serviceBindingProvider = serviceBindingProviders.get(notificationType); + if (serviceBindingProvider == null) { + serviceBindingProvider = + new NotificationServiceBindingProvider(binding, component, service, servletHost, + notificationTypeManager, encodingRegistry, httpUrl, this); + serviceBindingProviders.put(notificationType, serviceBindingProvider); + } + return serviceBindingProvider; + } + + private boolean validServiceBinding(NotificationBinding binding) { + URI notificationType = binding.getNotificationType(); + NotificationReferenceBindingProvider referenceBindingProvider = referenceBindingProviders.get(notificationType); + if (referenceBindingProvider != null) { + return validBinding(binding, referenceBindingProvider.getBinding()); + } + return true; + } + + private boolean validReferenceBinding(NotificationBinding binding) { + URI notificationType = binding.getNotificationType(); + NotificationServiceBindingProvider serviceBindingProvider = serviceBindingProviders.get(notificationType); + if (serviceBindingProvider != null) { + return validBinding(binding, serviceBindingProvider.getBinding()); + } + return true; + } + + private boolean validBinding(NotificationBinding binding1, NotificationBinding binding2) { + String binding1NtmAddress = binding1.getNtmAddress(); + String binding2NtmAddress = binding2.getNtmAddress(); + if (binding1NtmAddress == null && binding2NtmAddress == null) { + return true; + } else if (binding1NtmAddress == null || binding2NtmAddress == null) { + return false; + } else { + return binding1NtmAddress.equals(binding2NtmAddress); + } + } + + /* + * These methods are intended to be called by the binding providers' start + * methods. By the time this happens, both referenceBindingProvider != null && + * serviceBindingProvider != null, if they are ever going to be + */ + public void serviceProviderStarted(URI notificationType, + NotificationServiceBindingProvider serviceBindingProvider, + URL remoteNtmUrl) { + NotificationReferenceBindingProvider referenceBindingProvider = referenceBindingProviders.get(notificationType); + if (referenceBindingProvider == null) { + serviceBindingProvider.deployConsumer(); + } else if (referenceBindingProvider.isStarted()) { + String brokerID = BrokerID.generate(); + deployBroker(notificationType, serviceBindingProvider, referenceBindingProvider, brokerID, remoteNtmUrl); + } + } + + public void referenceProviderStarted(URI notificationType, + NotificationReferenceBindingProvider referenceBindingProvider, + URL remoteNtmUrl) { + NotificationServiceBindingProvider serviceBindingProvider = serviceBindingProviders.get(notificationType); + if (serviceBindingProvider == null) { + referenceBindingProvider.deployProducer(); + } else if (serviceBindingProvider.isStarted()) { + String brokerID = BrokerID.generate(); + deployBroker(notificationType, serviceBindingProvider, referenceBindingProvider, brokerID, remoteNtmUrl); + } + } + + private void deployBroker(URI notificationType, + NotificationServiceBindingProvider serviceBindingProvider, + NotificationReferenceBindingProvider referenceBindingProvider, + String brokerID, + URL remoteNtmUrl) { + URL consumerUrl = serviceBindingProvider.getURL(); + URL producerUrl = referenceBindingProvider.getURL(); + List<EndpointReference> consumerList = new ArrayList<EndpointReference>(); + List<EndpointReference> producerList = new ArrayList<EndpointReference>(); + boolean firstBroker = + notificationTypeManager.newBroker(notificationType, + consumerUrl, + producerUrl, + brokerID, + remoteNtmUrl, + consumerList, + producerList); + if (firstBroker) { + serviceBindingProvider.deployBroker(brokerID, null, producerList); + referenceBindingProvider.deployBroker(brokerID, null, consumerList); + if (!consumerList.isEmpty() || !producerList.isEmpty()) { + notificationTypeManager.newBrokerAck(remoteNtmUrl); + } + } else { + // returned lists contain broker consumers and producers and are the + // same length + int index = consumerList.size() - 1; + // establish connection with picked broker + EndpointReference brokerConsumerEPR = consumerList.get(index); + EndpointReference brokerProducerEPR = producerList.get(index); + serviceBindingProvider.deployBroker(brokerID, brokerProducerEPR, null); + referenceBindingProvider.deployBroker(brokerID, brokerConsumerEPR, null); + } + } + + public void replaceConsumersBrokerConnection(URI notificationType, EndpointReference chosenBrokerProducerEpr) { + NotificationServiceBindingProvider serviceBindingProvider = serviceBindingProviders.get(notificationType); + if (serviceBindingProvider == null) { + throw new RuntimeException("Missing service binding provider for [" + notificationType + "]"); + } + serviceBindingProvider.replaceBrokerConnection(chosenBrokerProducerEpr); + } + + public static void removeBroker(URI notificationType) { + if (factoryInstance == null) { + throw new RuntimeException("Missing factory instance"); + } + NotificationReferenceBindingProvider referenceBindingProvider = + factoryInstance.referenceBindingProviders.get(notificationType); + NotificationServiceBindingProvider serviceBindingProvider = + factoryInstance.serviceBindingProviders.get(notificationType); + if (referenceBindingProvider == null || serviceBindingProvider == null) { + throw new RuntimeException("Not a broker for [" + notificationType + "]"); + } + referenceBindingProvider.undeployBroker(serviceBindingProvider.getURL()); + } + + private String getBaseURI() { + if (httpUrl == null) { + String httpPort = System.getProperty("notification.httpPort"); + if (httpPort == null) { + httpPort = DEFAULT_PORT; + } + String localHost = null; + try { + localHost = InetAddress.getLocalHost().getCanonicalHostName(); + } catch (Exception e) { + e.printStackTrace(); + localHost = "localhost"; + } + httpUrl = "http://" + localHost + ((httpPort != null) ? (":" + httpPort) : ""); + } + return httpUrl; + } + + @SuppressWarnings("unchecked") + public void init() { + if (notificationTypeManager == null) { + AbstractEnDeCoder subscribeEnDeCoder = new SubscribeEnDeCoder(encodingRegistry); + subscribeEnDeCoder.start(); + AbstractEnDeCoder consumerReferenceEnDeCoder = new ConsumerReferenceEnDeCoder(encodingRegistry); + consumerReferenceEnDeCoder.start(); + AbstractEnDeCoder endpointAddressEnDeCoder = new EndpointAddressEnDeCoder(encodingRegistry); + endpointAddressEnDeCoder.start(); + AbstractEnDeCoder newConsumerEnDeCoder = new NewConsumerEnDeCoder(encodingRegistry); + newConsumerEnDeCoder.start(); + AbstractEnDeCoder newProducerEnDeCoder = new NewProducerEnDeCoder(encodingRegistry); + newProducerEnDeCoder.start(); + AbstractEnDeCoder newConsumerResponseEnDeCoder = new NewConsumerResponseEnDeCoder(encodingRegistry); + newConsumerResponseEnDeCoder.start(); + AbstractEnDeCoder newProducerResponseEnDeCoder = new NewProducerResponseEnDeCoder(encodingRegistry); + newProducerResponseEnDeCoder.start(); + AbstractEnDeCoder newBrokerEnDeCoder = new NewBrokerEnDeCoder(encodingRegistry); + newBrokerEnDeCoder.start(); + AbstractEnDeCoder brokerConsumerReferenceEnDeCoder = new BrokerConsumerReferenceEnDeCoder(encodingRegistry); + brokerConsumerReferenceEnDeCoder.start(); + AbstractEnDeCoder brokerProducerReferenceEnDeCoder = new BrokerProducerReferenceEnDeCoder(encodingRegistry); + brokerProducerReferenceEnDeCoder.start(); + AbstractEnDeCoder newBrokerResponseEnDeCoder = new NewBrokerResponseEnDeCoder(encodingRegistry); + newBrokerResponseEnDeCoder.start(); + AbstractEnDeCoder brokersEnDeCoder = new BrokersEnDeCoder(encodingRegistry); + brokersEnDeCoder.start(); + AbstractEnDeCoder brokerEnDeCoder = new BrokerEnDeCoder(encodingRegistry); + brokerEnDeCoder.start(); + AbstractEnDeCoder endConsumersEnDeCoder = new EndConsumersEnDeCoder(encodingRegistry); + endConsumersEnDeCoder.start(); + AbstractEnDeCoder endProducersEnDeCoder = new EndProducersEnDeCoder(encodingRegistry); + endProducersEnDeCoder.start(); + AbstractEnDeCoder endpointReferenceEnDeCoder = new EndpointReferenceEnDeCoder(encodingRegistry); + endpointReferenceEnDeCoder.start(); + AbstractEnDeCoder referencePropertiesEnDeCoder = new ReferencePropertiesEnDeCoder(encodingRegistry); + referencePropertiesEnDeCoder.start(); + AbstractEnDeCoder brokerIDEnDeCoder = new BrokerIDEnDeCoder(encodingRegistry); + brokerIDEnDeCoder.start(); + AbstractEnDeCoder connectionOverrideEnDeCoder = new ConnectionOverrideEnDeCoder(encodingRegistry); + connectionOverrideEnDeCoder.start(); + AbstractEnDeCoder connectionOverrideResponseEnDeCoder = + new ConnectionOverrideResponseEnDeCoder(encodingRegistry); + connectionOverrideResponseEnDeCoder.start(); + AbstractEnDeCoder newBrokerAckEnDeCoder = new NewBrokerAckEnDeCoder(encodingRegistry); + newBrokerAckEnDeCoder.start(); + AbstractEnDeCoder neighborBrokerConsumersEnDeCoder = new NeighborBrokerConsumersEnDeCoder(encodingRegistry); + neighborBrokerConsumersEnDeCoder.start(); + AbstractEnDeCoder removeBrokerEnDeCoder = new RemoveBrokerEnDeCoder(encodingRegistry); + removeBrokerEnDeCoder.start(); + AbstractEnDeCoder removedBrokerEnDeCoder = new RemovedBrokerEnDeCoder(encodingRegistry); + removedBrokerEnDeCoder.start(); + AbstractEnDeCoder neighborsEnDeCoder = new NeighborsEnDeCoder(encodingRegistry); + neighborsEnDeCoder.start(); + AbstractEnDeCoder replaceBrokerConnectionEnDeCoder = new ReplaceBrokerConnectionEnDeCoder(encodingRegistry); + replaceBrokerConnectionEnDeCoder.start(); + + notificationTypeManager = new NotificationTypeManagerImpl(); + notificationTypeManager.setServletHost(servletHost); + notificationTypeManager.setEncodingRegistry(encodingRegistry); + notificationTypeManager.init(); + + getBaseURI(); + } + } + +} |