001 /*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements. See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License. You may obtain a copy of the License at
008 *
009 * http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017 package org.apache.servicemix.common.endpoints;
018
019 import javax.jbi.component.ComponentContext;
020 import javax.jbi.messaging.ExchangeStatus;
021 import javax.jbi.messaging.InOnly;
022 import javax.jbi.messaging.MessageExchange;
023 import javax.jbi.messaging.NormalizedMessage;
024 import javax.jbi.messaging.RobustInOnly;
025 import javax.jbi.messaging.MessageExchange.Role;
026 import javax.jbi.servicedesc.ServiceEndpoint;
027 import javax.xml.namespace.QName;
028
029 import org.apache.servicemix.common.JbiConstants;
030 import org.apache.servicemix.common.DefaultComponent;
031 import org.apache.servicemix.common.ServiceUnit;
032
033 public abstract class ProviderEndpoint extends SimpleEndpoint {
034
035 private ServiceEndpoint activated;
036
037
038 public ProviderEndpoint() {
039 }
040
041 public ProviderEndpoint(ServiceUnit serviceUnit, QName service, String endpoint) {
042 super(serviceUnit, service, endpoint);
043 }
044
045 public ProviderEndpoint(DefaultComponent component, ServiceEndpoint endpoint) {
046 super(component.getServiceUnit(), endpoint.getServiceName(), endpoint.getEndpointName());
047 }
048
049 /* (non-Javadoc)
050 * @see org.apache.servicemix.common.Endpoint#getRole()
051 */
052 public Role getRole() {
053 return Role.PROVIDER;
054 }
055
056 public void activate() throws Exception {
057 super.activate();
058 ComponentContext ctx = getServiceUnit().getComponent().getComponentContext();
059 activated = ctx.activateEndpoint(service, endpoint);
060 }
061
062 public void deactivate() throws Exception {
063 if (activated == null) {
064 throw new IllegalStateException("Endpoint not activated: " + this);
065 }
066 ServiceEndpoint ep = activated;
067 activated = null;
068 ComponentContext ctx = getServiceUnit().getComponent().getComponentContext();
069 ctx.deactivateEndpoint(ep);
070 super.deactivate();
071 }
072
073 /**
074 * A default implementation of the message processor which checks the status of the exchange
075 * and if its valid will dispatch to either {@link #processInOnly(MessageExchange,NormalizedMessage)} for
076 * an {@link InOnly} or {@link RobustInOnly} message exchange otherwise the
077 * {@link #processInOut(MessageExchange,NormalizedMessage,NormalizedMessage)}
078 * method will be invoked
079 *
080 * @param exchange the message exchange
081 * @throws Exception
082 */
083 public void process(MessageExchange exchange) throws Exception {
084 // The component acts as a provider, this means that another component has requested our service
085 // As this exchange is active, this is either an in or a fault (out are sent by this component)
086 if (exchange.getRole() == Role.PROVIDER) {
087 // Exchange is finished
088 if (exchange.getStatus() == ExchangeStatus.DONE) {
089 return;
090 // Exchange has been aborted with an exception
091 } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
092 return;
093 // Exchange is active
094 } else {
095 NormalizedMessage in;
096 // Fault message
097 if (exchange.getFault() != null) {
098 done(exchange);
099 // In message
100 } else if ((in = exchange.getMessage("in")) != null) {
101 if (exchange instanceof InOnly || exchange instanceof RobustInOnly) {
102 processInOnly(exchange, in);
103 done(exchange);
104 }
105 else {
106 NormalizedMessage out = exchange.getMessage("out");
107 if (out == null) {
108 out = exchange.createMessage();
109 exchange.setMessage(out, "out");
110 }
111 processInOut(exchange, in, out);
112 boolean txSync = exchange.isTransacted() && Boolean.TRUE.equals(exchange.getProperty(JbiConstants.SEND_SYNC));
113 if (txSync) {
114 sendSync(exchange);
115 } else {
116 send(exchange);
117 }
118 }
119 // This is not compliant with the default MEPs
120 } else {
121 throw new IllegalStateException("Provider exchange is ACTIVE, but no in or fault is provided");
122 }
123 }
124 // Unsupported role: this should never happen has we never create exchanges
125 } else {
126 throw new IllegalStateException("Unsupported role: " + exchange.getRole());
127 }
128 }
129
130
131 protected void processInOnly(MessageExchange exchange, NormalizedMessage in) throws Exception {
132 throw new UnsupportedOperationException("Unsupported MEP: " + exchange.getPattern());
133 }
134
135 protected void processInOut(MessageExchange exchange, NormalizedMessage in, NormalizedMessage out) throws Exception {
136 throw new UnsupportedOperationException("Unsupported MEP: " + exchange.getPattern());
137 }
138
139 }