You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stratos.apache.org by im...@apache.org on 2013/11/04 12:14:09 UTC
[2/7] Refactored org.apache.stratos.lb.endpoint to
org.apache.stratos.load.balancer and org.apache.stratos.lb.common to
org.apache.stratos.load.balancer.common
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerContext.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerContext.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerContext.java
new file mode 100644
index 0000000..02f1a23
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/LoadBalancerContext.java
@@ -0,0 +1,181 @@
+/*
+ * 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.stratos.load.balancer;
+
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.engine.AxisConfiguration;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.synapse.config.SynapseConfiguration;
+import org.wso2.carbon.mediation.dependency.mgt.services.DependencyManagementService;
+import org.wso2.carbon.mediation.initializer.services.SynapseEnvironmentService;
+import org.wso2.carbon.registry.core.session.UserRegistry;
+import org.wso2.carbon.user.core.service.RealmService;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Defines load balancer context information.
+ */
+public class LoadBalancerContext {
+
+ private static volatile LoadBalancerContext instance;
+ private static final Log log = LogFactory.getLog(LoadBalancerContext.class);
+
+ private SynapseConfiguration synapseConfiguration;
+ private ConfigurationContext configCtxt;
+ private AxisConfiguration axisConfiguration;
+ private UserRegistry configRegistry;
+ private UserRegistry governanceRegistry;
+ private DependencyManagementService dependencyManager;
+ private Map<Integer, SynapseEnvironmentService> synapseEnvironmentServices;
+
+ private Map<String, ServiceContext> serviceContextMap;
+ private Map<String, ClusterContext> clusterContextMap;
+
+ private LoadBalancerContext() {
+ synapseEnvironmentServices = new HashMap<Integer, SynapseEnvironmentService>();
+
+ serviceContextMap = new HashMap<String, ServiceContext>();
+ clusterContextMap = new HashMap<String, ClusterContext>();
+ }
+
+ public static synchronized LoadBalancerContext getInstance() {
+ if (instance == null) {
+ synchronized (LoadBalancerContext.class){
+ if (instance == null) {
+ instance = new LoadBalancerContext ();
+ }
+ }
+ }
+ return instance;
+ }
+
+ public RealmService getRealmService() {
+ return realmService;
+ }
+
+ public void setRealmService(RealmService realmService) {
+ this.realmService = realmService;
+ }
+
+ private RealmService realmService;
+
+ public SynapseConfiguration getSynapseConfiguration() throws TenantAwareLoadBalanceEndpointException{
+ assertNull("SynapseConfiguration", synapseConfiguration);
+ return synapseConfiguration;
+ }
+
+ public void setSynapseConfiguration(SynapseConfiguration synapseConfiguration) {
+ this.synapseConfiguration = synapseConfiguration;
+ }
+
+ public AxisConfiguration getAxisConfiguration() throws TenantAwareLoadBalanceEndpointException {
+ assertNull("AxisConfiguration", axisConfiguration);
+ return axisConfiguration;
+ }
+
+ public void setAxisConfiguration(AxisConfiguration axisConfiguration) {
+ this.axisConfiguration = axisConfiguration;
+ }
+
+ public UserRegistry getConfigRegistry() throws TenantAwareLoadBalanceEndpointException {
+ assertNull("Registry", configRegistry);
+ return configRegistry;
+ }
+
+ public void setConfigRegistry(UserRegistry configRegistry) {
+ this.configRegistry = configRegistry;
+ }
+
+ public DependencyManagementService getDependencyManager() {
+ return dependencyManager;
+ }
+
+ public void setDependencyManager(DependencyManagementService dependencyManager) {
+ this.dependencyManager = dependencyManager;
+ }
+
+ private void assertNull(String name, Object object) throws TenantAwareLoadBalanceEndpointException {
+ if (object == null) {
+ String message = name + " reference in the proxy admin config holder is null";
+ log.error(message);
+ throw new TenantAwareLoadBalanceEndpointException(message);
+ }
+ }
+
+ public UserRegistry getGovernanceRegistry() {
+ return governanceRegistry;
+ }
+
+ public void setGovernanceRegistry(UserRegistry governanceRegistry) {
+ this.governanceRegistry = governanceRegistry;
+ }
+
+ public SynapseEnvironmentService getSynapseEnvironmentService(int id) {
+ return synapseEnvironmentServices.get(id);
+ }
+
+ public void addSynapseEnvironmentService(int id, SynapseEnvironmentService synapseEnvironmentService) {
+ synapseEnvironmentServices.put(id, synapseEnvironmentService);
+ }
+
+ public void removeSynapseEnvironmentService(int id) {
+ synapseEnvironmentServices.remove(id);
+ }
+
+ public Map<Integer, SynapseEnvironmentService> getSynapseEnvironmentServices() {
+ return synapseEnvironmentServices;
+ }
+
+ public ConfigurationContext getConfigCtxt() {
+ return configCtxt;
+ }
+
+ public void setConfigCtxt(ConfigurationContext configCtxt) {
+ this.configCtxt = configCtxt;
+ }
+
+ public Collection<ServiceContext> getServiceContexts() {
+ return serviceContextMap.values();
+ }
+
+ public ServiceContext getServiceContext(String serviceName) {
+ return serviceContextMap.get(serviceName);
+ }
+
+ public void addServiceContext(ServiceContext serviceContext) {
+ serviceContextMap.put(serviceContext.getServiceName(), serviceContext);
+ }
+
+ public Collection<ClusterContext> getClusterContexts() {
+ return clusterContextMap.values();
+ }
+
+ public ClusterContext getClusterContext(String clusterId) {
+ return clusterContextMap.get(clusterId);
+ }
+
+ public void addClusterContext(ClusterContext clusterContext) {
+ clusterContextMap.put(clusterContext.getClusterId(), clusterContext);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/RequestDelegator.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/RequestDelegator.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/RequestDelegator.java
new file mode 100644
index 0000000..583541b
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/RequestDelegator.java
@@ -0,0 +1,112 @@
+/*
+ * 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.stratos.load.balancer;
+
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.load.balancer.algorithm.AlgorithmContext;
+import org.apache.stratos.load.balancer.algorithm.LoadBalanceAlgorithm;
+import org.apache.stratos.messaging.message.receiver.topology.TopologyManager;
+import org.apache.stratos.messaging.domain.topology.Cluster;
+import org.apache.stratos.messaging.domain.topology.Member;
+import org.apache.stratos.messaging.domain.topology.Service;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+/**
+ * Implements core load balancing logic for identifying the next member
+ * according to the incoming request information.
+ */
+public class RequestDelegator {
+ private static final Log log = LogFactory.getLog(RequestDelegator.class);
+
+ private LoadBalanceAlgorithm algorithm;
+
+ public RequestDelegator(LoadBalanceAlgorithm algorithm) {
+ this.algorithm = algorithm;
+ }
+
+ public Member findNextMember(String targetHost) {
+
+ try {
+ if(targetHost == null)
+ return null;
+
+ TopologyManager.acquireReadLock();
+
+ Cluster cluster = findCluster(targetHost);
+ if(cluster != null) {
+ // Find algorithm context of the cluster
+ ClusterContext clusterContext = LoadBalancerContext.getInstance().getClusterContext(cluster.getClusterId());
+ if(clusterContext == null) {
+ clusterContext = new ClusterContext(cluster.getServiceName(), cluster.getClusterId());
+ LoadBalancerContext.getInstance().addClusterContext(clusterContext);
+ }
+
+ AlgorithmContext algorithmContext = clusterContext.getAlgorithmContext();
+ if(algorithmContext == null) {
+ algorithmContext = new AlgorithmContext(cluster.getServiceName(), cluster.getClusterId());
+ clusterContext.setAlgorithmContext(algorithmContext);
+ }
+ algorithm.setMembers(new ArrayList<Member>(cluster.getMembers()));
+ Member member = algorithm.getNextMember(algorithmContext);
+ if(log.isDebugEnabled()) {
+ log.debug(String.format("Next member found: service: %s cluster id: %s member id: %s", member.getServiceName(), member.getClusterId(), member.getMemberId()));
+ }
+ return member;
+ }
+ return null;
+ }
+ finally {
+ TopologyManager.releaseReadLock();
+ }
+ }
+
+ public Member findNextMember(String serviceName, int tenantId, String targetHost) {
+ throw new NotImplementedException();
+ }
+
+ private Service findService(String serviceName) {
+ Collection<Service> services = TopologyManager.getTopology().getServices();
+ for (Service service : services) {
+ if(service.getServiceName().equals(serviceName))
+ return service;
+ }
+ return null;
+ }
+
+ private Cluster findCluster(String targetHost) {
+ Collection<Service> services = TopologyManager.getTopology().getServices();
+ for (Service service : services) {
+ for (Cluster cluster : service.getClusters()) {
+ if (targetHost.equals(cluster.getHostName())) {
+ return cluster;
+ }
+ }
+ }
+ return null;
+ }
+
+ public boolean isTargetHostValid(String targetHost) {
+ return (findCluster(targetHost) != null);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/ServiceContext.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/ServiceContext.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/ServiceContext.java
new file mode 100644
index 0000000..58d461a
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/ServiceContext.java
@@ -0,0 +1,56 @@
+/*
+ * 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.stratos.load.balancer;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Defines service context properties.
+ */
+public class ServiceContext {
+ private String serviceName;
+ private Map<String, ClusterContext> clusterContextMap;
+
+ public ServiceContext() {
+ clusterContextMap = new HashMap<String, ClusterContext>();
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public void setServiceName(String serviceName) {
+ this.serviceName = serviceName;
+ }
+
+ public Collection<ClusterContext> getClusterContexts() {
+ return clusterContextMap.values();
+ }
+
+ public ClusterContext getClusterContext(String clusterId) {
+ return clusterContextMap.get(clusterId);
+ }
+
+ public void addClusterContext(ClusterContext clusterContext) {
+ clusterContextMap.put(clusterContext.getClusterId(), clusterContext);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/TenantAwareLoadBalanceEndpointException.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/TenantAwareLoadBalanceEndpointException.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/TenantAwareLoadBalanceEndpointException.java
new file mode 100644
index 0000000..de08988
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/TenantAwareLoadBalanceEndpointException.java
@@ -0,0 +1,39 @@
+/*
+ * 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.stratos.load.balancer;
+
+/**
+ * Exception to be thrown from this component.
+ */
+public class TenantAwareLoadBalanceEndpointException extends RuntimeException {
+
+ private static final long serialVersionUID = -663839410798538370L;
+
+ public TenantAwareLoadBalanceEndpointException(String msg) {
+ super(msg);
+ }
+
+ public TenantAwareLoadBalanceEndpointException(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+
+ public TenantAwareLoadBalanceEndpointException(Throwable cause) {
+ super(cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/AlgorithmContext.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/AlgorithmContext.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/AlgorithmContext.java
new file mode 100755
index 0000000..24bc70d
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/AlgorithmContext.java
@@ -0,0 +1,51 @@
+/*
+ * 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.stratos.load.balancer.algorithm;
+
+/**
+ * Algorithm context is used for identifying the cluster and its current member for executing load balancing algorithms.
+ */
+public class AlgorithmContext {
+ private String serviceName;
+ private String clusterId;
+ private int currentMemberIndex;
+
+ public AlgorithmContext(String serviceName, String clusterId) {
+ this.serviceName = serviceName;
+ this.clusterId = clusterId;
+ this.currentMemberIndex = 0;
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public String getClusterId() {
+ return clusterId;
+ }
+
+ public int getCurrentMemberIndex() {
+ return currentMemberIndex;
+ }
+
+ public void setCurrentMemberIndex(int currentMemberIndex) {
+ this.currentMemberIndex = currentMemberIndex;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithm.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithm.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithm.java
new file mode 100755
index 0000000..55412cf
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithm.java
@@ -0,0 +1,65 @@
+/*
+ * 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.stratos.load.balancer.algorithm;
+
+import org.apache.stratos.messaging.domain.topology.Member;
+
+import java.util.List;
+
+/**
+ * Defines the specification for implementing load balancing algorithms.
+ */
+public interface LoadBalanceAlgorithm {
+ /**
+ * Return algorithm name.
+ *
+ * @return
+ */
+ public String getName();
+
+ /**
+ * Apply the algorithm and return the next member.
+ *
+ * @param algorithmContext
+ * @return
+ */
+ public Member getNextMember(AlgorithmContext algorithmContext);
+
+ /**
+ * Set member list of a given cluster.
+ *
+ * @param members
+ */
+ public void setMembers(List<Member> members);
+
+ /**
+ * Reset the algorithm and start from the beginning.
+ *
+ * @param algorithmContext
+ */
+ public void reset(AlgorithmContext algorithmContext);
+
+ /**
+ * Clone algorithm object.
+ *
+ * @return
+ */
+ public LoadBalanceAlgorithm clone();
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithmFactory.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithmFactory.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithmFactory.java
new file mode 100644
index 0000000..699741e
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/LoadBalanceAlgorithmFactory.java
@@ -0,0 +1,69 @@
+/*
+ * 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.stratos.load.balancer.algorithm;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.lang.reflect.InvocationTargetException;
+
+/**
+ * Load balance algorithm factory to create the algorithm
+ */
+public class LoadBalanceAlgorithmFactory {
+ private static final Log log = LogFactory.getLog(LoadBalanceAlgorithmFactory.class);
+
+
+ public static LoadBalanceAlgorithm createAlgorithm(String className) {
+ try {
+ Class algorithmClass = Class.forName(className);
+ try {
+ Object instance = algorithmClass.getConstructor().newInstance();
+ if (instance instanceof LoadBalanceAlgorithm) {
+ return (LoadBalanceAlgorithm) instance;
+ } else {
+ throw new RuntimeException(String.format("Class %s is not a valid load balance algorithm implementation"));
+ }
+ } catch (NoSuchMethodException e) {
+ if (log.isErrorEnabled()) {
+ log.error(e);
+ }
+ } catch (InstantiationException e) {
+ if (log.isErrorEnabled()) {
+ log.error(e);
+ }
+ } catch (IllegalAccessException e) {
+ if (log.isErrorEnabled()) {
+ log.error(e);
+ }
+ } catch (InvocationTargetException e) {
+ if (log.isErrorEnabled()) {
+ log.error(e);
+ }
+ }
+
+ } catch (ClassNotFoundException e) {
+ if (log.isErrorEnabled()) {
+ log.error(e);
+ }
+ }
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/RoundRobin.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/RoundRobin.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/RoundRobin.java
new file mode 100644
index 0000000..e3c2898
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/algorithm/RoundRobin.java
@@ -0,0 +1,97 @@
+/*
+ * 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.stratos.load.balancer.algorithm;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.messaging.domain.topology.Member;
+
+import java.util.List;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * This is the implementation of the round robin load balancing algorithm. It simply iterates
+ * through the endpoint list one by one for until an active endpoint is found.
+ */
+public class RoundRobin implements LoadBalanceAlgorithm {
+ private static final Log log = LogFactory.getLog(RoundRobin.class);
+
+ private List<Member> members;
+ private final Lock lock = new ReentrantLock();
+
+ @Override
+ public String getName() {
+ return "Round Robin";
+ }
+
+ @Override
+ public Member getNextMember(AlgorithmContext algorithmContext) {
+ if (members.size() == 0) {
+ return null;
+ }
+ Member current = null;
+ lock.lock();
+ try {
+ int currentMemberIndex = algorithmContext.getCurrentMemberIndex();
+ if (currentMemberIndex >= members.size()) {
+ currentMemberIndex = 0;
+ }
+ int index = members.size();
+ do { current = members.get(currentMemberIndex);
+ if (currentMemberIndex == members.size() - 1) {
+ currentMemberIndex = 0;
+ } else {
+ currentMemberIndex++;
+ }
+ index--;
+ } while ((!current.isActive()) && index > 0);
+ algorithmContext.setCurrentMemberIndex(currentMemberIndex);
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Service name: %s cluster id: %s members: %d", algorithmContext.getServiceName(), algorithmContext.getClusterId(), members.size()));
+ log.debug(String.format("Current member: %s", current.getMemberId()));
+ }
+
+ } finally {
+ lock.unlock();
+ }
+ return current;
+ }
+
+ @Override
+ public void setMembers(List<Member> members) {
+ this.members = members;
+ }
+
+ @Override
+ public void reset(AlgorithmContext algorithmContext) {
+ if (log.isDebugEnabled()) {
+ log.debug("Resetting the Round Robin load balancing algorithm ...");
+ }
+ synchronized (algorithmContext) {
+ algorithmContext.setCurrentMemberIndex(0);
+ }
+ }
+
+ @Override
+ public LoadBalanceAlgorithm clone() {
+ return new RoundRobin();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/TenantAwareLoadBalanceEndpoint.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/TenantAwareLoadBalanceEndpoint.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/TenantAwareLoadBalanceEndpoint.java
new file mode 100644
index 0000000..a132b30
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/endpoint/TenantAwareLoadBalanceEndpoint.java
@@ -0,0 +1,517 @@
+/*
+ * 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.stratos.load.balancer.endpoint;
+
+import org.apache.axis2.addressing.EndpointReference;
+import org.apache.axis2.description.TransportInDescription;
+import org.apache.http.protocol.HTTP;
+import org.apache.stratos.load.balancer.RequestDelegator;
+import org.apache.stratos.load.balancer.algorithm.LoadBalanceAlgorithmFactory;
+import org.apache.stratos.load.balancer.stat.LoadBalancingStatsCollector;
+import org.apache.stratos.load.balancer.util.Constants;
+import org.apache.stratos.messaging.domain.topology.Member;
+import org.apache.stratos.messaging.domain.topology.Port;
+import org.apache.synapse.MessageContext;
+import org.apache.synapse.SynapseConstants;
+import org.apache.synapse.SynapseException;
+import org.apache.synapse.core.SynapseEnvironment;
+import org.apache.synapse.core.axis2.Axis2MessageContext;
+import org.apache.synapse.endpoints.AddressEndpoint;
+import org.apache.synapse.endpoints.DynamicLoadbalanceFaultHandler;
+import org.apache.synapse.endpoints.Endpoint;
+import org.apache.synapse.endpoints.EndpointDefinition;
+import org.apache.synapse.endpoints.dispatch.HttpSessionDispatcher;
+import org.apache.synapse.endpoints.dispatch.SessionInformation;
+import org.apache.synapse.transport.nhttp.NhttpConstants;
+
+import java.io.Serializable;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.*;
+
+
+public class TenantAwareLoadBalanceEndpoint extends org.apache.synapse.endpoints.LoadbalanceEndpoint implements Serializable {
+ private static final String PORT_MAPPING_PREFIX = "port.mapping.";
+
+ /* Request delegator identifies the next member */
+ private RequestDelegator requestDelegator;
+
+ /* Load balance algorithm class name */
+ private String algorithmClassName;
+
+ /* Flag to enable session affinity based load balancing */
+ private boolean sessionAffinity;
+
+ /* Dispatcher used for session affinity */
+ private HttpSessionDispatcher dispatcher;
+
+ /* Sessions time out interval */
+ private long sessionTimeout = -1;
+
+ @Override
+ public void init(SynapseEnvironment synapseEnvironment) {
+ super.init(synapseEnvironment);
+
+ requestDelegator = new RequestDelegator(LoadBalanceAlgorithmFactory.createAlgorithm(algorithmClassName));
+ synapseEnvironment.getSynapseConfiguration().setProperty(SynapseConstants.PROP_SAL_ENDPOINT_DEFAULT_SESSION_TIMEOUT, String.valueOf(sessionTimeout));
+ setDispatcher(new HttpSessionDispatcher());
+ }
+
+ @Override
+ public void send(MessageContext synCtx) {
+
+ SessionInformation sessionInformation = null;
+ org.apache.axis2.clustering.Member currentMember = null;
+ if (isSessionAffinityBasedLB()) {
+ // Check existing session information
+ sessionInformation = (SessionInformation) synCtx.getProperty(
+ SynapseConstants.PROP_SAL_CURRENT_SESSION_INFORMATION);
+
+ currentMember = (org.apache.axis2.clustering.Member) synCtx.getProperty(
+ SynapseConstants.PROP_SAL_ENDPOINT_CURRENT_MEMBER);
+
+ if (sessionInformation == null && currentMember == null) {
+ sessionInformation = dispatcher.getSession(synCtx);
+ if (sessionInformation != null) {
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Existing session found: %s", sessionInformation.getId()));
+ }
+
+ currentMember = sessionInformation.getMember();
+ synCtx.setProperty(SynapseConstants.PROP_SAL_ENDPOINT_CURRENT_MEMBER, currentMember);
+ // This is for reliably recovery any session information if while response is getting ,
+ // session information has been removed by cleaner.
+ // This will not be a cost as session information is not a heavy data structure
+ synCtx.setProperty(SynapseConstants.PROP_SAL_CURRENT_SESSION_INFORMATION, sessionInformation);
+ }
+ }
+
+ }
+
+ TenantAwareLoadBalanceFaultHandler faultHandler = new TenantAwareLoadBalanceFaultHandler();
+ if (sessionInformation != null && currentMember != null) {
+ // Send request to the member with the existing session
+ sessionInformation.updateExpiryTime();
+ sendToApplicationMember(synCtx, currentMember, faultHandler, false);
+ } else {
+ // No existing session found
+ // Find next member
+ org.apache.axis2.clustering.Member axis2Member = findNextMember(synCtx);
+ if (axis2Member != null) {
+ // Send request to member
+ sendToApplicationMember(synCtx, axis2Member, faultHandler, true);
+ } else {
+ throwSynapseException(synCtx, 404, "Active application instances not found");
+ }
+ }
+ }
+
+ private void throwSynapseException(MessageContext synCtx, int errorCode, String errorMessage) {
+ synCtx.setProperty(SynapseConstants.ERROR_CODE, errorCode);
+ synCtx.setProperty(SynapseConstants.ERROR_MESSAGE, errorMessage);
+ throw new SynapseException(errorMessage);
+ }
+
+ /**
+ * Setup load balancer message context properties to be used by the out block of the main sequence.
+ * These values will be used to update the Location value in the response header.
+ *
+ * @param synCtx
+ */
+ private void setupLoadBalancerContextProperties(MessageContext synCtx) {
+ String lbHostName = extractTargetHost(synCtx);
+ org.apache.axis2.context.MessageContext axis2MsgCtx = ((Axis2MessageContext) synCtx).getAxis2MessageContext();
+ TransportInDescription httpTransportIn = axis2MsgCtx.getConfigurationContext().getAxisConfiguration().getTransportIn("http");
+ TransportInDescription httpsTransportIn = axis2MsgCtx.getConfigurationContext().getAxisConfiguration().getTransportIn("https");
+ String lbHttpPort = (String) httpTransportIn.getParameter("port").getValue();
+ String lbHttpsPort = (String) httpsTransportIn.getParameter("port").getValue();
+
+ synCtx.setProperty(Constants.LB_HOST_NAME, lbHostName);
+ synCtx.setProperty(Constants.LB_HTTP_PORT, lbHttpPort);
+ synCtx.setProperty(Constants.LB_HTTPS_PORT, lbHttpsPort);
+ }
+
+
+ /**
+ * Adding the X-Forwarded-For/X-Originating-IP headers to the outgoing message.
+ *
+ * @param synCtx Current message context
+ */
+ protected void setupTransportHeaders(MessageContext synCtx) {
+ Axis2MessageContext axis2smc = (Axis2MessageContext) synCtx;
+ org.apache.axis2.context.MessageContext axis2MessageCtx = axis2smc.getAxis2MessageContext();
+ Object headers = axis2MessageCtx.getProperty(org.apache.axis2.context.MessageContext.TRANSPORT_HEADERS);
+ if (headers != null && headers instanceof Map) {
+ Map headersMap = (Map) headers;
+ String xForwardFor = (String) headersMap.get(NhttpConstants.HEADER_X_FORWARDED_FOR);
+ String remoteHost = (String) axis2MessageCtx.getProperty(org.apache.axis2.context.MessageContext.REMOTE_ADDR);
+
+ if (xForwardFor != null && !"".equals(xForwardFor)) {
+ StringBuilder xForwardedForString = new StringBuilder();
+ xForwardedForString.append(xForwardFor);
+ if (remoteHost != null && !"".equals(remoteHost)) {
+ xForwardedForString.append(",").append(remoteHost);
+ }
+ headersMap.put(NhttpConstants.HEADER_X_FORWARDED_FOR, xForwardedForString.toString());
+ } else {
+ headersMap.put(NhttpConstants.HEADER_X_FORWARDED_FOR, remoteHost);
+ }
+
+ //Extracting information of X-Originating-IP
+ if (headersMap.get(NhttpConstants.HEADER_X_ORIGINATING_IP_FORM_1) != null) {
+ headersMap.put(NhttpConstants.HEADER_X_ORIGINATING_IP_FORM_1, headersMap.get(NhttpConstants.HEADER_X_ORIGINATING_IP_FORM_1));
+ } else if (headersMap.get(NhttpConstants.HEADER_X_ORIGINATING_IP_FORM_2) != null) {
+ headersMap.put(NhttpConstants.HEADER_X_ORIGINATING_IP_FORM_2, headersMap.get(NhttpConstants.HEADER_X_ORIGINATING_IP_FORM_2));
+ }
+
+ }
+ }
+
+ private org.apache.axis2.clustering.Member findNextMember(MessageContext synCtx) {
+ String targetHost = extractTargetHost(synCtx);
+ if(!requestDelegator.isTargetHostValid(targetHost)) {
+ throwSynapseException(synCtx, 404, String.format("Unknown host name %s", targetHost));
+ }
+ Member member = requestDelegator.findNextMember(targetHost);
+ if (member == null)
+ return null;
+
+ // Create Axi2 member object
+ String transport = extractTransport(synCtx);
+ Port transportPort = member.getPort(transport);
+ if (transportPort == null) {
+ if(log.isErrorEnabled()) {
+ log.error(String.format("Port not found for transport %s in member %s", transport, member.getMemberId()));
+ }
+ throwSynapseException(synCtx, 500, "Internal server error");
+ }
+
+ int memberPort = transportPort.getValue();
+ org.apache.axis2.clustering.Member axis2Member = new org.apache.axis2.clustering.Member(member.getMemberIp(), memberPort);
+ axis2Member.setDomain(member.getClusterId());
+ Port httpPort = member.getPort("http");
+ if (httpPort != null)
+ axis2Member.setHttpPort(httpPort.getValue());
+ Port httpsPort = member.getPort("https");
+ if (httpsPort != null)
+ axis2Member.setHttpsPort(httpsPort.getValue());
+ axis2Member.setActive(member.isActive());
+ return axis2Member;
+ }
+
+ private String extractTargetHost(MessageContext synCtx) {
+ org.apache.axis2.context.MessageContext msgCtx =
+ ((Axis2MessageContext) synCtx).getAxis2MessageContext();
+
+ Map headerMap = (Map) msgCtx.getProperty(org.apache.axis2.context.MessageContext.TRANSPORT_HEADERS);
+ String hostName = null;
+ if (headerMap != null) {
+ Object hostObj = headerMap.get(HTTP.TARGET_HOST);
+ hostName = (String) hostObj;
+ if (hostName.contains(":")) {
+ hostName = hostName.substring(0, hostName.indexOf(":"));
+ }
+ }
+ return hostName;
+ }
+
+ private int extractPort(MessageContext synCtx, String transport) {
+ org.apache.axis2.context.MessageContext msgCtx =
+ ((Axis2MessageContext) synCtx).getAxis2MessageContext();
+
+ Map headerMap = (Map) msgCtx.getProperty(org.apache.axis2.context.MessageContext.TRANSPORT_HEADERS);
+ int port = -1;
+ if (headerMap != null) {
+ String hostHeader = (String) headerMap.get(HTTP.TARGET_HOST);
+ int index = hostHeader.indexOf(':');
+ if (index != -1) {
+ port = Integer.parseInt(hostHeader.trim().substring(index + 1));
+ } else {
+ if ("http".equals(transport)) {
+ port = 80;
+ } else if ("https".equals(transport)) {
+ port = 443;
+ }
+ }
+ }
+ return port;
+ }
+
+ private String extractTransport(MessageContext synCtx) {
+ org.apache.axis2.context.MessageContext axis2MessageContext = ((Axis2MessageContext) synCtx).getAxis2MessageContext();
+ return axis2MessageContext.getTransportIn().getName();
+ }
+
+ /**
+ * @param to get an endpoint to send the information
+ * @param member The member to which an EP has to be created
+ * @param synCtx synapse context
+ * @return the created endpoint
+ */
+ private Endpoint getEndpoint(EndpointReference to, org.apache.axis2.clustering.Member member, MessageContext synCtx) {
+ AddressEndpoint endpoint = new AddressEndpoint();
+ endpoint.setEnableMBeanStats(false);
+ endpoint.setName("DLB:" + member.getHostName() +
+ ":" + member.getPort() + ":" + UUID.randomUUID());
+ EndpointDefinition definition = new EndpointDefinition();
+ definition.setSuspendMaximumDuration(10000);
+ definition.setReplicationDisabled(true);
+ definition.setAddress(to.getAddress());
+ endpoint.setDefinition(definition);
+ endpoint.init((SynapseEnvironment)
+ ((Axis2MessageContext) synCtx).getAxis2MessageContext().
+ getConfigurationContext().getAxisConfiguration().
+ getParameterValue(SynapseConstants.SYNAPSE_ENV));
+ return endpoint;
+ }
+
+ private EndpointReference getEndpointReferenceAfterURLRewrite(org.apache.axis2.clustering.Member currentMember,
+ String transport,
+ String address,
+ int incomingPort) {
+
+ if (transport.startsWith("https")) {
+ transport = "https";
+ } else if (transport.startsWith("http")) {
+ transport = "http";
+ } else {
+ String msg = "Cannot load balance for non-HTTP/S transport " + transport;
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+ // URL Rewrite
+ if (transport.startsWith("http") || transport.startsWith("https")) {
+ if (address.startsWith("http://") || address.startsWith("https://")) {
+ try {
+ String _address = address.indexOf("?") > 0 ? address.substring(address.indexOf("?"), address.length()) : "";
+ address = new URL(address).getPath() + _address;
+ } catch (MalformedURLException e) {
+ String msg = "URL " + address + " is malformed";
+ log.error(msg, e);
+ throw new SynapseException(msg, e);
+ }
+ }
+
+ int port;
+ Properties memberProperties = currentMember.getProperties();
+ String mappedPort = memberProperties.getProperty(PORT_MAPPING_PREFIX + incomingPort);
+ if (mappedPort != null) {
+ port = Integer.parseInt(mappedPort);
+ } else if (transport.startsWith("https")) {
+ port = currentMember.getHttpsPort();
+ } else {
+ port = currentMember.getHttpPort();
+ }
+
+ String remoteHost = memberProperties.getProperty("remoteHost");
+ String hostName = (remoteHost == null) ? currentMember.getHostName() : remoteHost;
+ return new EndpointReference(transport + "://" + hostName +
+ ":" + port + address);
+ } else {
+ String msg = "Cannot load balance for non-HTTP/S transport " + transport;
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+ }
+
+ /*
+ * Preparing the endpoint sequence for a new session establishment request
+ */
+ private void prepareEndPointSequence(MessageContext synCtx, Endpoint endpoint) {
+
+ Object o = synCtx.getProperty(SynapseConstants.PROP_SAL_ENDPOINT_ENDPOINT_LIST);
+ List<Endpoint> endpointList;
+ if (o instanceof List) {
+ endpointList = (List<Endpoint>) o;
+ endpointList.add(this);
+
+ } else {
+ // this is the first endpoint in the hierarchy. so create the queue and
+ // insert this as the first element.
+ endpointList = new ArrayList<Endpoint>();
+ endpointList.add(this);
+ synCtx.setProperty(SynapseConstants.PROP_SAL_ENDPOINT_ENDPOINT_LIST, endpointList);
+ }
+
+ // if the next endpoint is not a session affinity one, endpoint sequence ends
+ // here. but we have to add the next endpoint to the list.
+ if (!(endpoint instanceof TenantAwareLoadBalanceEndpoint)) {
+ endpointList.add(endpoint);
+ // Clearing out if there any any session information with current message
+ if (dispatcher.isServerInitiatedSession()) {
+ dispatcher.removeSessionID(synCtx);
+ }
+ }
+ }
+
+ protected void sendToApplicationMember(MessageContext synCtx,
+ org.apache.axis2.clustering.Member currentMember,
+ DynamicLoadbalanceFaultHandler faultHandler,
+ boolean newSession) {
+ //Rewriting the URL
+ org.apache.axis2.context.MessageContext axis2MsgCtx =
+ ((Axis2MessageContext) synCtx).getAxis2MessageContext();
+
+ //Removing the REST_URL_POSTFIX - this is a hack.
+ //In this load balance endpoint we create an endpoint per request by setting the complete url as the address.
+ //If a REST message comes Axis2FlexibleMEPClient append the REST_URL_POSTFIX to the address. Hence endpoint fails
+ //do send the request. e.g. http://localhost:8080/example/index.html/example/index.html
+ axis2MsgCtx.removeProperty(NhttpConstants.REST_URL_POSTFIX);
+
+ String transport = axis2MsgCtx.getTransportIn().getName();
+ String address = synCtx.getTo().getAddress();
+ int incomingPort = extractPort(synCtx, transport);
+ EndpointReference to = getEndpointReferenceAfterURLRewrite(currentMember, transport, address, incomingPort);
+ synCtx.setTo(to);
+
+ faultHandler.setTo(to);
+ faultHandler.setCurrentMember(currentMember);
+ synCtx.pushFaultHandler(faultHandler);
+ if (isFailover()) {
+ synCtx.getEnvelope().build();
+ }
+
+ Endpoint endpoint = getEndpoint(to, currentMember, synCtx);
+ faultHandler.setCurrentEp(endpoint);
+ if (isSessionAffinityBasedLB() && newSession) {
+ prepareEndPointSequence(synCtx, endpoint);
+ synCtx.setProperty(SynapseConstants.PROP_SAL_ENDPOINT_CURRENT_MEMBER, currentMember);
+ synCtx.setProperty(SynapseConstants.PROP_SAL_ENDPOINT_CURRENT_DISPATCHER, dispatcher);
+ // we should also indicate that this is the first message in the session. so that
+ // onFault(...) method can resend only the failed attempts for the first message.
+ synCtx.setProperty(SynapseConstants.PROP_SAL_ENDPOINT_FIRST_MESSAGE_IN_SESSION,
+ Boolean.TRUE);
+ }
+
+ Map<String, String> memberHosts;
+ if ((memberHosts = (Map<String, String>) currentMember.getProperties().get(HttpSessionDispatcher.HOSTS)) == null) {
+ currentMember.getProperties().put(HttpSessionDispatcher.HOSTS,
+ memberHosts = new HashMap<String, String>());
+ }
+ memberHosts.put(extractTargetHost(synCtx), "true");
+ setupTransportHeaders(synCtx);
+ setupLoadBalancerContextProperties(synCtx);
+
+ // Update health stats
+ LoadBalancingStatsCollector.getInstance().incrementRequestInflightCount(currentMember.getDomain());
+ // Set the cluster id in the message context
+ synCtx.setProperty(Constants.CLUSTER_ID, currentMember.getDomain());
+
+ try {
+ if (log.isDebugEnabled()) {
+ log.debug(String.format("Sending request to endpoint: %s", to.getAddress()));
+ }
+ endpoint.send(synCtx);
+ } catch (Exception e) {
+ if (e.getMessage().toLowerCase().contains("io reactor shutdown")) {
+ log.fatal("System cannot continue normal operation. Restarting", e);
+ System.exit(121); // restart
+ } else {
+ throw new SynapseException(e);
+ }
+ }
+ }
+
+ public void setDispatcher(HttpSessionDispatcher dispatcher) {
+ this.dispatcher = dispatcher;
+ }
+
+ public HttpSessionDispatcher getDispatcher() {
+ return dispatcher;
+ }
+
+ public String getAlgorithmClassName() {
+ return algorithmClassName;
+ }
+
+ public void setAlgorithmClassName(String algorithmClassName) {
+ this.algorithmClassName = algorithmClassName;
+ }
+
+ public long getSessionTimeout() {
+ return sessionTimeout;
+ }
+
+ public void setSessionTimeout(long sessionTimeout) {
+ this.sessionTimeout = sessionTimeout;
+ }
+
+ public boolean isSessionAffinityBasedLB() {
+ return sessionAffinity;
+ }
+
+ public void setSessionAffinity(boolean sessionAffinity) {
+ this.sessionAffinity = sessionAffinity;
+ }
+
+ private class TenantAwareLoadBalanceFaultHandler extends DynamicLoadbalanceFaultHandler {
+ private org.apache.axis2.clustering.Member currentMember;
+ private Endpoint currentEp;
+ private EndpointReference to;
+
+ @Override
+ public void setCurrentMember(org.apache.axis2.clustering.Member currentMember) {
+ this.currentMember = currentMember;
+ }
+
+ @Override
+ public void setCurrentEp(Endpoint currentEp) {
+ this.currentEp = currentEp;
+ }
+
+ @Override
+ public void setTo(EndpointReference to) {
+ this.to = to;
+ }
+
+ @Override
+ public void onFault(MessageContext synCtx) {
+ //cleanup endpoint if exists
+ if (currentEp != null) {
+ currentEp.destroy();
+ }
+ if (currentMember == null) {
+ return;
+ }
+
+ Stack faultStack = synCtx.getFaultStack();
+ if (faultStack != null && !faultStack.isEmpty()) {
+ faultStack.pop(); // Remove the LoadbalanceFaultHandler
+ }
+
+ currentMember = findNextMember(synCtx);
+ if (currentMember == null) {
+ String msg = String.format("No application members available to serve the request %s", synCtx.getTo().getAddress());
+ log.error(msg);
+ throw new SynapseException(msg);
+ }
+ synCtx.setTo(to);
+ if (isSessionAffinityBasedLB()) {
+ //We are sending the this message on a new session,
+ // hence we need to remove previous session information
+ Set pros = synCtx.getPropertyKeySet();
+ if (pros != null) {
+ pros.remove(SynapseConstants.PROP_SAL_CURRENT_SESSION_INFORMATION);
+ }
+ }
+ sendToApplicationMember(synCtx, currentMember, this, true);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/LoadBalanceEndpointServiceComponent.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/LoadBalanceEndpointServiceComponent.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/LoadBalanceEndpointServiceComponent.java
new file mode 100644
index 0000000..b317d78
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/internal/LoadBalanceEndpointServiceComponent.java
@@ -0,0 +1,333 @@
+/*
+ * 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.stratos.load.balancer.internal;
+
+import org.apache.axis2.deployment.DeploymentEngine;
+import org.apache.axis2.engine.AxisConfiguration;
+import org.apache.stratos.load.balancer.LoadBalancerContext;
+import org.apache.stratos.load.balancer.TenantAwareLoadBalanceEndpointException;
+import org.apache.stratos.messaging.message.receiver.topology.TopologyEventMessageDelegator;
+import org.apache.stratos.messaging.message.receiver.topology.TopologyEventMessageReceiver;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.messaging.broker.subscribe.TopicSubscriber;
+import org.apache.stratos.messaging.util.Constants;
+import org.apache.synapse.config.SynapseConfiguration;
+import org.apache.synapse.config.xml.MultiXMLConfigurationBuilder;
+import org.apache.synapse.core.SynapseEnvironment;
+import org.apache.synapse.deployers.SynapseArtifactDeploymentStore;
+import org.apache.synapse.endpoints.Endpoint;
+import org.osgi.service.component.ComponentContext;
+import org.wso2.carbon.mediation.dependency.mgt.services.DependencyManagementService;
+import org.wso2.carbon.mediation.initializer.ServiceBusConstants;
+import org.wso2.carbon.mediation.initializer.ServiceBusUtils;
+import org.wso2.carbon.mediation.initializer.services.SynapseConfigurationService;
+import org.wso2.carbon.mediation.initializer.services.SynapseEnvironmentService;
+import org.wso2.carbon.mediation.initializer.services.SynapseRegistrationsService;
+import org.wso2.carbon.registry.core.exceptions.RegistryException;
+import org.wso2.carbon.registry.core.service.RegistryService;
+import org.wso2.carbon.utils.ConfigurationContextService;
+import org.wso2.carbon.utils.multitenancy.MultitenantConstants;
+import org.wso2.carbon.user.core.service.RealmService;
+import org.apache.stratos.load.balancer.EndpointDeployer;
+
+import java.io.File;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * @scr.component name="org.apache.stratos.lbr.endpoint" immediate="true"
+ * @scr.reference name="configuration.context.service"
+ * interface="org.wso2.carbon.utils.ConfigurationContextService"
+ * cardinality="1..1" policy="dynamic"
+ * bind="setConfigurationContextService"
+ * unbind="unsetConfigurationContextService"
+ * @scr.reference name="synapse.config.service" interface=
+ * "org.wso2.carbon.mediation.initializer.services.SynapseConfigurationService"
+ * cardinality="1..1" policy="dynamic"
+ * bind="setSynapseConfigurationService"
+ * unbind="unsetSynapseConfigurationService"
+ * @scr.reference name="synapse.env.service" interface=
+ * "org.wso2.carbon.mediation.initializer.services.SynapseEnvironmentService"
+ * cardinality="1..n" policy="dynamic"
+ * bind="setSynapseEnvironmentService"
+ * unbind="unsetSynapseEnvironmentService"
+ * @scr.reference name="registry.service"
+ * interface="org.wso2.carbon.registry.core.service.RegistryService"
+ * cardinality="1..1" policy="dynamic" bind="setRegistryService"
+ * unbind="unsetRegistryService"
+ * @scr.reference name="dependency.mgt.service" interface=
+ * "org.wso2.carbon.mediation.dependency.mgt.services.DependencyManagementService"
+ * cardinality="0..1" policy="dynamic"
+ * bind="setDependencyManager" unbind="unsetDependencyManager"
+ * @scr.reference name="synapse.registrations.service" interface=
+ * "org.wso2.carbon.mediation.initializer.services.SynapseRegistrationsService"
+ * cardinality="1..n" policy="dynamic"
+ * bind="setSynapseRegistrationsService"
+ * unbind="unsetSynapseRegistrationsService"
+ * @scr.reference name="user.realmservice.default"
+ * interface="org.wso2.carbon.user.core.service.RealmService"
+ * cardinality="1..1" policy="dynamic" bind="setRealmService"
+ * unbind="unsetRealmService"
+ */
+@SuppressWarnings({"UnusedDeclaration", "JavaDoc"})
+public class LoadBalanceEndpointServiceComponent {
+ private static final Log log = LogFactory.getLog(LoadBalanceEndpointServiceComponent.class);
+
+ private boolean activated = false;
+
+ protected void activate(ComponentContext ctxt) {
+ try {
+ // Register endpoint deployer
+ SynapseEnvironmentService synEnvService = LoadBalancerContext.getInstance().getSynapseEnvironmentService(
+ MultitenantConstants.SUPER_TENANT_ID);
+ registerDeployer(LoadBalancerContext.getInstance().getAxisConfiguration(),
+ synEnvService.getSynapseEnvironment());
+
+ // Start topic subscriber thread
+ TopicSubscriber topicSubscriber = new TopicSubscriber(Constants.TOPOLOGY_TOPIC);
+ topicSubscriber.setMessageListener(new TopologyEventMessageReceiver());
+ Thread subscriberThread = new Thread(topicSubscriber);
+ subscriberThread.start();
+ if (log.isDebugEnabled()) {
+ log.debug("Topology event message receiver thread started");
+ }
+
+ // Start topology message receiver thread
+ Thread receiverThread = new Thread(new TopologyEventMessageDelegator());
+ receiverThread.start();
+ if (log.isDebugEnabled()) {
+ log.debug("Topology message processor thread started");
+ }
+
+ activated = true;
+ if (log.isDebugEnabled()) {
+ log.debug("LoadBalanceEndpointServiceComponent is activated ");
+ }
+ } catch (Throwable e) {
+ log.error("Failed to activate LoadBalanceEndpointServiceComponent", e);
+ }
+ }
+
+ protected void deactivate(ComponentContext context) {
+ try {
+ Set<Map.Entry<Integer, SynapseEnvironmentService>> entrySet = LoadBalancerContext
+ .getInstance().getSynapseEnvironmentServices().entrySet();
+ for (Map.Entry<Integer, SynapseEnvironmentService> entry : entrySet) {
+ unregisterDeployer(entry.getValue().getConfigurationContext()
+ .getAxisConfiguration(), entry.getValue()
+ .getSynapseEnvironment());
+ }
+ } catch (Exception e) {
+ log.warn("Couldn't remove the EndpointDeployer");
+ }
+ }
+
+ /**
+ * Un-registers the Endpoint deployer.
+ *
+ * @param axisConfig AxisConfiguration to which this deployer belongs
+ * @param synapseEnvironment SynapseEnvironment to which this deployer belongs
+ */
+ private void unregisterDeployer(AxisConfiguration axisConfig,
+ SynapseEnvironment synapseEnvironment)
+ throws TenantAwareLoadBalanceEndpointException {
+ if (axisConfig != null) {
+ DeploymentEngine deploymentEngine = (DeploymentEngine) axisConfig
+ .getConfigurator();
+ String synapseConfigPath = ServiceBusUtils
+ .getSynapseConfigAbsPath(synapseEnvironment
+ .getServerContextInformation());
+ String endpointDirPath = synapseConfigPath + File.separator
+ + MultiXMLConfigurationBuilder.ENDPOINTS_DIR;
+ deploymentEngine.removeDeployer(endpointDirPath,
+ ServiceBusConstants.ARTIFACT_EXTENSION);
+ }
+ }
+
+ /**
+ * Registers the Endpoint deployer.
+ *
+ * @param axisConfig AxisConfiguration to which this deployer belongs
+ * @param synapseEnvironment SynapseEnvironment to which this deployer belongs
+ */
+ private void registerDeployer(AxisConfiguration axisConfig,
+ SynapseEnvironment synapseEnvironment)
+ throws TenantAwareLoadBalanceEndpointException {
+ SynapseConfiguration synCfg = synapseEnvironment
+ .getSynapseConfiguration();
+ DeploymentEngine deploymentEngine = (DeploymentEngine) axisConfig
+ .getConfigurator();
+ SynapseArtifactDeploymentStore deploymentStore = synCfg
+ .getArtifactDeploymentStore();
+
+ String synapseConfigPath = ServiceBusUtils
+ .getSynapseConfigAbsPath(synapseEnvironment
+ .getServerContextInformation());
+ String endpointDirPath = synapseConfigPath + File.separator
+ + MultiXMLConfigurationBuilder.ENDPOINTS_DIR;
+
+ for (Endpoint ep : synCfg.getDefinedEndpoints().values()) {
+ if (ep.getFileName() != null) {
+ deploymentStore.addRestoredArtifact(endpointDirPath
+ + File.separator + ep.getFileName());
+ }
+ }
+ deploymentEngine.addDeployer(new EndpointDeployer(), endpointDirPath,
+ ServiceBusConstants.ARTIFACT_EXTENSION);
+ }
+
+ protected void setConfigurationContextService(ConfigurationContextService cfgCtxService) {
+ LoadBalancerContext.getInstance().setAxisConfiguration(cfgCtxService.getServerConfigContext().getAxisConfiguration());
+ LoadBalancerContext.getInstance().setConfigCtxt(cfgCtxService.getServerConfigContext());
+ }
+
+ protected void unsetConfigurationContextService(ConfigurationContextService cfgCtxService) {
+ LoadBalancerContext.getInstance().setAxisConfiguration(null);
+ LoadBalancerContext.getInstance().setConfigCtxt(null);
+ }
+
+ protected void setSynapseConfigurationService(SynapseConfigurationService synapseConfigurationService) {
+ LoadBalancerContext.getInstance().setSynapseConfiguration(synapseConfigurationService.getSynapseConfiguration());
+ }
+
+ protected void unsetSynapseConfigurationService(SynapseConfigurationService synapseConfigurationService) {
+ LoadBalancerContext.getInstance().setSynapseConfiguration(null);
+ }
+
+ /**
+ * Here we receive an event about the creation of a SynapseEnvironment. If
+ * this is SuperTenant we have to wait until all the other constraints are
+ * met and actual initialization is done in the activate method. Otherwise
+ * we have to do the activation here.
+ *
+ * @param synapseEnvironmentService SynapseEnvironmentService which contains information about the
+ * new Synapse Instance
+ */
+ protected void setSynapseEnvironmentService(SynapseEnvironmentService synapseEnvironmentService) {
+ boolean alreadyCreated = LoadBalancerContext.getInstance()
+ .getSynapseEnvironmentServices()
+ .containsKey(synapseEnvironmentService.getTenantId());
+
+ LoadBalancerContext.getInstance().addSynapseEnvironmentService(
+ synapseEnvironmentService.getTenantId(),
+ synapseEnvironmentService);
+ if (activated) {
+ if (!alreadyCreated) {
+ try {
+ registerDeployer(synapseEnvironmentService
+ .getConfigurationContext().getAxisConfiguration(),
+ synapseEnvironmentService.getSynapseEnvironment());
+ if (log.isDebugEnabled()) {
+ log.debug("Endpoint Admin bundle is activated ");
+ }
+ } catch (Throwable e) {
+ log.error("Failed to activate Endpoint Admin bundle ", e);
+ }
+ }
+ }
+ }
+
+ /**
+ * Here we receive an event about Destroying a SynapseEnvironment. This can
+ * be the super tenant destruction or a tenant destruction.
+ *
+ * @param synapseEnvironmentService synapseEnvironment
+ */
+ protected void unsetSynapseEnvironmentService(
+ SynapseEnvironmentService synapseEnvironmentService) {
+ LoadBalancerContext.getInstance().removeSynapseEnvironmentService(
+ synapseEnvironmentService.getTenantId());
+ }
+
+ protected void setRegistryService(RegistryService regService) {
+ if (log.isDebugEnabled()) {
+ log.debug("RegistryService bound to the endpoint component");
+ }
+ try {
+ LoadBalancerContext.getInstance().setConfigRegistry(
+ regService.getConfigSystemRegistry());
+ LoadBalancerContext.getInstance().setGovernanceRegistry(
+ regService.getGovernanceSystemRegistry());
+ } catch (RegistryException e) {
+ log.error("Couldn't retrieve the registry from the registry service");
+ }
+ }
+
+ protected void unsetRegistryService(RegistryService regService) {
+ if (log.isDebugEnabled()) {
+ log.debug("RegistryService unbound from the endpoint component");
+ }
+ LoadBalancerContext.getInstance().setConfigRegistry(null);
+ }
+
+ protected void setDependencyManager(
+ DependencyManagementService dependencyMgr) {
+ if (log.isDebugEnabled()) {
+ log.debug("Dependency management service bound to the endpoint component");
+ }
+ LoadBalancerContext.getInstance().setDependencyManager(dependencyMgr);
+ }
+
+ protected void unsetDependencyManager(
+ DependencyManagementService dependencyMgr) {
+ if (log.isDebugEnabled()) {
+ log.debug("Dependency management service unbound from the endpoint component");
+ }
+ LoadBalancerContext.getInstance().setDependencyManager(null);
+ }
+
+ protected void setSynapseRegistrationsService(
+ SynapseRegistrationsService synapseRegistrationsService) {
+
+ }
+
+ protected void unsetSynapseRegistrationsService(
+ SynapseRegistrationsService synapseRegistrationsService) {
+ int tenantId = synapseRegistrationsService.getTenantId();
+ if (LoadBalancerContext.getInstance().getSynapseEnvironmentServices()
+ .containsKey(tenantId)) {
+ SynapseEnvironment env = LoadBalancerContext.getInstance()
+ .getSynapseEnvironmentService(tenantId)
+ .getSynapseEnvironment();
+
+ LoadBalancerContext.getInstance().removeSynapseEnvironmentService(
+ synapseRegistrationsService.getTenantId());
+
+ AxisConfiguration axisConfig = synapseRegistrationsService
+ .getConfigurationContext().getAxisConfiguration();
+ if (axisConfig != null) {
+ try {
+ unregisterDeployer(axisConfig, env);
+ } catch (Exception e) {
+ log.warn("Couldn't remove the EndpointDeployer");
+ }
+ }
+ }
+ }
+
+ protected void setRealmService(RealmService realmService) {
+ LoadBalancerContext.getInstance().setRealmService(realmService);
+ }
+
+ protected void unsetRealmService(RealmService realmService) {
+ LoadBalancerContext.getInstance().setRealmService(null);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/ResponseInterceptor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/ResponseInterceptor.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/ResponseInterceptor.java
new file mode 100644
index 0000000..fc6d84c
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/mediators/ResponseInterceptor.java
@@ -0,0 +1,58 @@
+/**
+ * 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.stratos.load.balancer.mediators;
+
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.stratos.load.balancer.stat.LoadBalancingStatsCollector;
+import org.apache.stratos.load.balancer.util.Constants;
+import org.apache.synapse.ManagedLifecycle;
+import org.apache.synapse.MessageContext;
+import org.apache.synapse.core.SynapseEnvironment;
+import org.apache.synapse.core.axis2.Axis2MessageContext;
+import org.apache.synapse.mediators.AbstractMediator;
+
+/**
+ * This Synapse mediator counts the responses that are going across LB.
+ *
+ */
+public class ResponseInterceptor extends AbstractMediator implements ManagedLifecycle {
+
+ public boolean mediate(MessageContext synCtx) {
+ if(log.isDebugEnabled()) {
+ log.debug("Mediation started " + ResponseInterceptor.class.getName());
+ }
+ String clusterId = (String) synCtx.getProperty(Constants.CLUSTER_ID);
+ LoadBalancingStatsCollector.getInstance().decrementRequestInflightCount(clusterId);
+ return true;
+ }
+
+ @Override
+ public void destroy() {
+ if(log.isDebugEnabled()) {
+ log.debug("ResponseInterceptor mediator destroyed");
+ }
+ }
+
+ @Override
+ public void init(SynapseEnvironment arg0) {
+ if(log.isDebugEnabled()) {
+ log.debug("ResponseInterceptor mediator initiated");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/stat/LoadBalancingStatsCollector.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/stat/LoadBalancingStatsCollector.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/stat/LoadBalancingStatsCollector.java
new file mode 100644
index 0000000..a0adc93
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/stat/LoadBalancingStatsCollector.java
@@ -0,0 +1,110 @@
+/*
+ * 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.stratos.load.balancer.stat;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Observable;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.stratos.load.balancer.stat.observers.WSO2CEPStatsObserver;
+
+/**
+ * This is the load balancing stats collector and any observer can get registered here
+ * and receive notifications periodically.
+ * This is a Singleton object.
+ * @author nirmal
+ *
+ */
+public class LoadBalancingStatsCollector extends Observable{
+
+ private static LoadBalancingStatsCollector collector;
+ private Map<String, Integer> clusterIdToRequestInflightCountMap;
+ private Thread notifier;
+
+ private LoadBalancingStatsCollector() {
+ clusterIdToRequestInflightCountMap = new ConcurrentHashMap<String, Integer>();
+ if (notifier == null || (notifier != null && !notifier.isAlive())) {
+ notifier = new Thread(new ObserverNotifier());
+ notifier.start();
+ }
+ }
+
+ public static LoadBalancingStatsCollector getInstance() {
+ if (collector == null) {
+ synchronized (LoadBalancingStatsCollector.class) {
+ if (collector == null) {
+ collector = new LoadBalancingStatsCollector();
+ // add observers
+ collector.addObserver(new WSO2CEPStatsObserver());
+ }
+ }
+ }
+ return collector;
+ }
+
+ public void incrementRequestInflightCount(String clusterId) {
+ if(clusterId == null) {
+ return;
+ }
+
+ int value = 1;
+ if(clusterIdToRequestInflightCountMap.get(clusterId) != null) {
+ value += clusterIdToRequestInflightCountMap.get(clusterId);
+ }
+ clusterIdToRequestInflightCountMap.put(clusterId, value);
+ setChanged();
+ }
+
+ public void decrementRequestInflightCount(String clusterId) {
+ if(clusterId == null) {
+ return;
+ }
+
+ int value = -1;
+ if(clusterIdToRequestInflightCountMap.get(clusterId) != null) {
+ value += clusterIdToRequestInflightCountMap.get(clusterId);
+ }
+ clusterIdToRequestInflightCountMap.put(clusterId, value);
+ setChanged();
+ }
+
+
+ /**
+ * This thread will notify all the observers of this subject.
+ * @author nirmal
+ *
+ */
+ private class ObserverNotifier implements Runnable {
+
+ @Override
+ public void run() {
+ while(true) {
+ try {
+ Thread.sleep(15000);
+ } catch (InterruptedException ignore) {
+ }
+ LoadBalancingStatsCollector.getInstance().notifyObservers(new HashMap<String, Integer>(clusterIdToRequestInflightCountMap));
+ }
+
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/stat/observers/WSO2CEPStatsObserver.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/stat/observers/WSO2CEPStatsObserver.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/stat/observers/WSO2CEPStatsObserver.java
new file mode 100644
index 0000000..6093ee5
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/stat/observers/WSO2CEPStatsObserver.java
@@ -0,0 +1,102 @@
+/*
+ * 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.stratos.load.balancer.stat.observers;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Observable;
+import java.util.Observer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.wso2.carbon.databridge.agent.thrift.Agent;
+import org.wso2.carbon.databridge.agent.thrift.AsyncDataPublisher;
+import org.wso2.carbon.databridge.agent.thrift.conf.AgentConfiguration;
+import org.wso2.carbon.databridge.agent.thrift.exception.AgentException;
+import org.wso2.carbon.databridge.commons.Event;
+import org.wso2.carbon.utils.CarbonUtils;
+
+public class WSO2CEPStatsObserver implements Observer{
+
+ private static final Log log = LogFactory.getLog(WSO2CEPStatsObserver.class);
+ private static final String CALL_CENTER_DATA_STREAM = "stratos.lb.stats";
+ private static final String VERSION = "1.0.0";
+ private AsyncDataPublisher asyncDataPublisher;
+
+ public WSO2CEPStatsObserver() {
+ AgentConfiguration agentConfiguration = new AgentConfiguration();
+ // TODO get following from somewhere, without hard-coding.
+ System.setProperty("javax.net.ssl.trustStore", CarbonUtils.getCarbonHome()+File.separator+"repository"+
+ File.separator+"resources"+File.separator+"security"+File.separator+"client-truststore.jks" );
+ System.setProperty("javax.net.ssl.trustStorePassword", "wso2carbon");
+
+ Agent agent = new Agent(agentConfiguration);
+ //TODO read following from a config file?
+ String ip = System.getProperty("thrift.receiver.ip");
+ String port = System.getProperty("thrift.receiver.port");
+ //Using Asynchronous data publisher
+ asyncDataPublisher = new AsyncDataPublisher("tcp://"+ip+":"+port+"", "admin", "admin", agent);
+ String streamDefinition = "{" +
+ " 'name':'" + CALL_CENTER_DATA_STREAM + "'," +
+ " 'version':'" + VERSION + "'," +
+ " 'nickName': 'lb stats'," +
+ " 'description': 'lb stats'," +
+ " 'metaData':[]," +
+ " 'payloadData':[" +
+ " {'name':'cluster_id','type':'STRING'}," +
+ " {'name':'in_flight_requests','type':'INT'}" +
+ " ]" +
+ "}";
+ asyncDataPublisher.addStreamDefinition(streamDefinition, CALL_CENTER_DATA_STREAM, VERSION);
+ }
+
+ public void update(Observable arg0, Object arg1) {
+ if(arg1 != null && arg1 instanceof Map<?, ?>) {
+ Map<String, Integer> stats = (Map<String, Integer>)arg1;
+ publishEvents(stats);
+ }
+ }
+
+ private void publishEvents(Map<String, Integer> stats) {
+
+ for (Map.Entry<String, Integer> entry : stats.entrySet()) {
+
+ Object[] payload = new Object[]{entry.getKey(), entry.getValue()};
+ Event event = eventObject(null, null, payload, new HashMap<String, String>());
+ try {
+ asyncDataPublisher.publish(CALL_CENTER_DATA_STREAM, VERSION, event);
+ } catch (AgentException e) {
+ log.error("Failed to publish events. ", e);
+ }
+
+ }
+ stats = null;
+ }
+
+ private static Event eventObject(Object[] correlationData, Object[] metaData,
+ Object[] payLoadData, HashMap<String, String> map) {
+ Event event = new Event();
+ event.setCorrelationData(correlationData);
+ event.setMetaData(metaData);
+ event.setPayloadData(payLoadData);
+ event.setArbitraryDataMap(map);
+ return event;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/Constants.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/Constants.java b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/Constants.java
new file mode 100644
index 0000000..bdbb5ac
--- /dev/null
+++ b/components/org.apache.stratos.load.balancer/src/main/java/org/apache/stratos/load/balancer/util/Constants.java
@@ -0,0 +1,27 @@
+/*
+ * 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.stratos.load.balancer.util;
+
+public class Constants {
+
+ public static final String CLUSTER_ID = "cluster_id";
+ public static final String LB_HOST_NAME = "LB_HOST_NAME";
+ public static final String LB_HTTP_PORT = "LB_HTTP_PORT";
+ public static final String LB_HTTPS_PORT = "LB_HTTPS_PORT";
+}
http://git-wip-us.apache.org/repos/asf/incubator-stratos/blob/b6793bdf/components/pom.xml
----------------------------------------------------------------------
diff --git a/components/pom.xml b/components/pom.xml
index 478663a..bd1ce2f 100644
--- a/components/pom.xml
+++ b/components/pom.xml
@@ -51,8 +51,8 @@
<!-- CLI -->
<module>org.apache.stratos.cli</module>
<!-- ELB -->
- <module>org.apache.stratos.lb.common</module>
- <module>org.apache.stratos.lb.endpoint</module>
+ <module>org.apache.stratos.load.balancer.common</module>
+ <module>org.apache.stratos.load.balancer</module>
<!-- Tenant Mgt -->
<module>org.apache.stratos.register.ui</module>
<module>org.apache.stratos.tenant.mgt.core</module>