001: /*
002: * Licensed to the Apache Software Foundation (ASF) under one
003: * or more contributor license agreements. See the NOTICE file
004: * distributed with this work for additional information
005: * regarding copyright ownership. The ASF licenses this file
006: * to you under the Apache License, Version 2.0 (the
007: * "License"); you may not use this file except in compliance
008: * with the License. You may obtain a copy of the License at
009: *
010: * http://www.apache.org/licenses/LICENSE-2.0
011: *
012: * Unless required by applicable law or agreed to in writing,
013: * software distributed under the License is distributed on an
014: * * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
015: * KIND, either express or implied. See the License for the
016: * specific language governing permissions and limitations
017: * under the License.
018: */
019:
020: package org.apache.synapse.endpoints;
021:
022: import org.apache.axis2.clustering.ClusterManager;
023: import org.apache.axis2.context.ConfigurationContext;
024: import org.apache.commons.logging.Log;
025: import org.apache.commons.logging.LogFactory;
026: import org.apache.synapse.FaultHandler;
027: import org.apache.synapse.MessageContext;
028: import org.apache.synapse.SynapseConstants;
029: import org.apache.synapse.core.axis2.Axis2MessageContext;
030: import org.apache.synapse.endpoints.algorithms.AlgorithmContext;
031: import org.apache.synapse.endpoints.algorithms.LoadbalanceAlgorithm;
032:
033: import java.util.List;
034:
035: /**
036: * Load balance endpoint can have multiple endpoints. It will route messages according to the
037: * specified load balancing algorithm. This will assume that all immediate child endpoints are identical
038: * in state (state is replicated) or state is not maintained at those endpoints. If an endpoint is
039: * failing, the failed endpoint is marked as inactive and the message to the next endpoint obtained
040: * using the load balancing algorithm. If all the endpoints have failed and the parent endpoint is
041: * available, onChildEndpointFail(...) method of parent endpoint is called. If parent is not
042: * available, this will call next FaultHandler for the message context.
043: */
044: public class LoadbalanceEndpoint implements Endpoint {
045: private static final Log log = LogFactory
046: .getLog(FailoverEndpoint.class);
047: /**
048: * Name of the endpoint. Used for named endpoints which can be referred using the key attribute
049: * of indirect endpoints.
050: */
051: private String name = null;
052:
053: /**
054: * List of endpoints among which the load is distributed. Any object implementing the Endpoint
055: * interface could be used.
056: */
057: private List endpoints = null;
058:
059: /**
060: * Algorithm used for selecting the next endpoint to direct the load. Default is RoundRobin.
061: */
062: private LoadbalanceAlgorithm algorithm = null;
063:
064: /**
065: * If this supports load balancing with failover. If true, request will be directed to the next
066: * endpoint if the current one is failing.
067: */
068: private boolean failover = true;
069:
070: /**
071: * Parent endpoint of this endpoint if this used inside another endpoint. Possible parents are
072: * LoadbalanceEndpoint, SALoadbalanceEndpoint and FailoverEndpoint objects.
073: */
074: private Endpoint parentEndpoint = null;
075:
076: /**
077: * The endpoint context , place holder for keep any runtime states related to the endpoint
078: */
079: private final EndpointContext endpointContext = new EndpointContext();
080:
081: /**
082: * The algorithm context , place holder for keep any runtime states related to the load balance
083: * algorithm
084: */
085: private final AlgorithmContext algorithmContext = new AlgorithmContext();
086:
087: public void send(MessageContext synMessageContext) {
088:
089: if (log.isDebugEnabled()) {
090: log.debug("Start : Load-balance Endpoint");
091: }
092:
093: boolean isClusteringEnable = false;
094: // get Axis2 MessageContext and ConfigurationContext
095: org.apache.axis2.context.MessageContext axisMC = ((Axis2MessageContext) synMessageContext)
096: .getAxis2MessageContext();
097: ConfigurationContext cc = axisMC.getConfigurationContext();
098:
099: //The check for clustering environment
100:
101: ClusterManager clusterManager = cc.getAxisConfiguration()
102: .getClusterManager();
103: if (clusterManager != null
104: && clusterManager.getContextManager() != null) {
105: isClusteringEnable = true;
106: }
107:
108: String endPointName = this .getName();
109: if (endPointName == null) {
110:
111: if (log.isDebugEnabled() && isClusteringEnable) {
112: log
113: .warn("In a clustering environment , the endpoint name should be specified"
114: + "even for anonymous endpoints. Otherwise , the clustering would not be "
115: + "functioned correctly if there are more than one anonymous endpoints. ");
116: }
117: endPointName = SynapseConstants.ANONYMOUS_ENDPOINT;
118: }
119:
120: if (isClusteringEnable) {
121:
122: // if this is a cluster environment , then set configuration context to endpoint context
123: if (endpointContext.getConfigurationContext() == null) {
124: endpointContext.setConfigurationContext(cc);
125: endpointContext.setContextID(endPointName);
126:
127: }
128: // if this is a cluster environment , then set configuration context to load balance
129: // algorithm context
130: if (algorithmContext.getConfigurationContext() == null) {
131: algorithmContext.setConfigurationContext(cc);
132: algorithmContext.setContextID(endPointName);
133: }
134: }
135:
136: Endpoint endpoint = algorithm.getNextEndpoint(
137: synMessageContext, algorithmContext);
138: if (endpoint != null) {
139:
140: // We have to build the envelop if we are supporting failover.
141: // Failover should sent the original message multiple times if failures occur. So we have to
142: // access the envelop multiple times.
143: if (failover) {
144: synMessageContext.getEnvelope().build();
145: }
146:
147: endpoint.send(synMessageContext);
148:
149: } else {
150: // there are no active child endpoints. so mark this endpoint as failed.
151: setActive(false, synMessageContext);
152:
153: if (parentEndpoint != null) {
154: parentEndpoint.onChildEndpointFail(this ,
155: synMessageContext);
156: } else {
157: Object o = synMessageContext.getFaultStack().pop();
158: if (o != null) {
159: ((FaultHandler) o).handleFault(synMessageContext);
160: }
161: }
162: }
163: }
164:
165: public String getName() {
166: return name;
167: }
168:
169: public void setName(String name) {
170: this .name = name.trim();
171: }
172:
173: public LoadbalanceAlgorithm getAlgorithm() {
174: return algorithm;
175: }
176:
177: public void setAlgorithm(LoadbalanceAlgorithm algorithm) {
178: this .algorithm = algorithm;
179: }
180:
181: /**
182: * If this endpoint is in inactive state, checks if all immediate child endpoints are still
183: * failed. If so returns false. If at least one child endpoint is in active state, sets this
184: * endpoint's state to active and returns true. As this a sessionless load balancing endpoint
185: * having one active child endpoint is enough to consider this as active.
186: *
187: * @param synMessageContext MessageContext of the current message. This is not used here.
188: * @return true if active. false otherwise.
189: */
190: public boolean isActive(MessageContext synMessageContext) {
191: boolean active = endpointContext.isActive();
192: if (!active && endpoints != null) {
193: for (int i = 0; i < endpoints.size(); i++) {
194: Endpoint endpoint = (Endpoint) endpoints.get(i);
195: if (endpoint.isActive(synMessageContext)) {
196: endpointContext.setActive(true);
197:
198: // don't break the loop though we found one active endpoint. calling isActive()
199: // on all child endpoints will update their active state. so this is a good
200: // time to do that.
201: }
202: }
203: }
204:
205: return active;
206: }
207:
208: public void setActive(boolean active,
209: MessageContext synMessageContext) {
210: // setting a volatile boolean variable is thread safe.
211: endpointContext.setActive(active);
212: }
213:
214: public boolean isFailover() {
215: return failover;
216: }
217:
218: public void setFailover(boolean failover) {
219: this .failover = failover;
220: }
221:
222: public List getEndpoints() {
223: return endpoints;
224: }
225:
226: public void setEndpoints(List endpoints) {
227: this .endpoints = endpoints;
228: }
229:
230: public void setParentEndpoint(Endpoint parentEndpoint) {
231: this .parentEndpoint = parentEndpoint;
232: }
233:
234: public void onChildEndpointFail(Endpoint endpoint,
235: MessageContext synMessageContext) {
236:
237: // resend (to a different endpoint) only if we support failover
238: if (failover) {
239: send(synMessageContext);
240: } else {
241: // we are not informing this to the parent endpoint as the failure of this loadbalance
242: // endpoint. there can be more active endpoints under this, and current request has
243: // failed only because the currently selected child endpoint has failed AND failover is
244: // turned off in this load balance endpoint. so just call the next fault handler.
245: Object o = synMessageContext.getFaultStack().pop();
246: if (o != null) {
247: ((FaultHandler) o).handleFault(synMessageContext);
248: }
249: }
250: }
251: }
|