CAMEL-8526: Add more EIP as specialized mbeans
Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/72a06ac3 Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/72a06ac3 Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/72a06ac3 Branch: refs/heads/master Commit: 72a06ac39632af22f66121384c6d991535cad7e3 Parents: 6f27b9a Author: Claus Ibsen <davscl...@apache.org> Authored: Fri Jul 24 13:06:10 2015 +0200 Committer: Claus Ibsen <davscl...@apache.org> Committed: Fri Jul 24 13:06:10 2015 +0200 ---------------------------------------------------------------------- .../mbean/ManagedTopicLoadBalancerMBean.java | 26 ++++++ .../mbean/ManagedWeightedBalancerMBean.java | 35 ++++++++ .../DefaultManagementObjectStrategy.java | 8 ++ .../mbean/ManagedTopicLoadBalancer.java | 41 +++++++++ .../mbean/ManagedWeightedLoadBalancer.java | 77 +++++++++++++++++ .../WeightedLoadBalancerDefinition.java | 8 +- .../ManagedTopicLoadBalancerTest.java | 81 ++++++++++++++++++ .../ManagedWeightedLoadBalancerTest.java | 90 ++++++++++++++++++++ 8 files changed, 360 insertions(+), 6 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/72a06ac3/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedTopicLoadBalancerMBean.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedTopicLoadBalancerMBean.java b/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedTopicLoadBalancerMBean.java new file mode 100644 index 0000000..0db4e75 --- /dev/null +++ b/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedTopicLoadBalancerMBean.java @@ -0,0 +1,26 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.api.management.mbean; + +import org.apache.camel.api.management.ManagedAttribute; + +public interface ManagedTopicLoadBalancerMBean extends ManagedProcessorMBean { + + @ManagedAttribute(description = "Number of processors in the load balancer") + Integer getSize(); + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/camel/blob/72a06ac3/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedWeightedBalancerMBean.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedWeightedBalancerMBean.java b/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedWeightedBalancerMBean.java new file mode 100644 index 0000000..2671126 --- /dev/null +++ b/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedWeightedBalancerMBean.java @@ -0,0 +1,35 @@ +/** + * 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.camel.api.management.mbean; + +import org.apache.camel.api.management.ManagedAttribute; + +public interface ManagedWeightedBalancerMBean extends ManagedProcessorMBean { + + @ManagedAttribute(description = "Number of processors in the load balancer") + Integer getSize(); + + @ManagedAttribute(description = "Whether round robin mode is enabled") + Boolean isRoundRobin(); + + @ManagedAttribute(description = "The distribution ratio of integer weights separated by delimiters for example '2,3,5'") + String getDistributionRatio(); + + @ManagedAttribute(description = "The delimiter") + String getDistributionRatioDelimiter(); + +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/camel/blob/72a06ac3/camel-core/src/main/java/org/apache/camel/management/DefaultManagementObjectStrategy.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/management/DefaultManagementObjectStrategy.java b/camel-core/src/main/java/org/apache/camel/management/DefaultManagementObjectStrategy.java index e4562af..27669f7 100644 --- a/camel-core/src/main/java/org/apache/camel/management/DefaultManagementObjectStrategy.java +++ b/camel-core/src/main/java/org/apache/camel/management/DefaultManagementObjectStrategy.java @@ -86,9 +86,11 @@ import org.apache.camel.management.mbean.ManagedThreads; import org.apache.camel.management.mbean.ManagedThrottler; import org.apache.camel.management.mbean.ManagedThroughputLogger; import org.apache.camel.management.mbean.ManagedThrowException; +import org.apache.camel.management.mbean.ManagedTopicLoadBalancer; import org.apache.camel.management.mbean.ManagedTransformer; import org.apache.camel.management.mbean.ManagedUnmarshal; import org.apache.camel.management.mbean.ManagedValidate; +import org.apache.camel.management.mbean.ManagedWeightedLoadBalancer; import org.apache.camel.management.mbean.ManagedWireTapProcessor; import org.apache.camel.model.ModelCamelContext; import org.apache.camel.model.ProcessDefinition; @@ -139,6 +141,8 @@ import org.apache.camel.processor.idempotent.IdempotentConsumer; import org.apache.camel.processor.loadbalancer.RandomLoadBalancer; import org.apache.camel.processor.loadbalancer.RoundRobinLoadBalancer; import org.apache.camel.processor.loadbalancer.StickyLoadBalancer; +import org.apache.camel.processor.loadbalancer.TopicLoadBalancer; +import org.apache.camel.processor.loadbalancer.WeightedLoadBalancer; import org.apache.camel.processor.validation.PredicateValidatingProcessor; import org.apache.camel.spi.BrowsableEndpoint; import org.apache.camel.spi.EventNotifier; @@ -297,6 +301,10 @@ public class DefaultManagementObjectStrategy implements ManagementObjectStrategy answer = new ManagedRoundRobinLoadBalancer(context, (RoundRobinLoadBalancer) target, (org.apache.camel.model.LoadBalanceDefinition) definition); } else if (target instanceof StickyLoadBalancer) { answer = new ManagedStickyLoadBalancer(context, (StickyLoadBalancer) target, (org.apache.camel.model.LoadBalanceDefinition) definition); + } else if (target instanceof TopicLoadBalancer) { + answer = new ManagedTopicLoadBalancer(context, (TopicLoadBalancer) target, (org.apache.camel.model.LoadBalanceDefinition) definition); + } else if (target instanceof WeightedLoadBalancer) { + answer = new ManagedWeightedLoadBalancer(context, (WeightedLoadBalancer) target, (org.apache.camel.model.LoadBalanceDefinition) definition); } else if (target instanceof RecipientList) { answer = new ManagedRecipientList(context, (RecipientList) target, (RecipientListDefinition) definition); } else if (target instanceof Splitter) { http://git-wip-us.apache.org/repos/asf/camel/blob/72a06ac3/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedTopicLoadBalancer.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedTopicLoadBalancer.java b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedTopicLoadBalancer.java new file mode 100644 index 0000000..ce5cc64 --- /dev/null +++ b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedTopicLoadBalancer.java @@ -0,0 +1,41 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.management.mbean; + +import org.apache.camel.CamelContext; +import org.apache.camel.api.management.ManagedResource; +import org.apache.camel.api.management.mbean.ManagedTopicLoadBalancerMBean; +import org.apache.camel.model.LoadBalanceDefinition; +import org.apache.camel.processor.loadbalancer.TopicLoadBalancer; + +/** + * @version + */ +@ManagedResource(description = "Managed Topic LoadBalancer") +public class ManagedTopicLoadBalancer extends ManagedProcessor implements ManagedTopicLoadBalancerMBean { + private final TopicLoadBalancer processor; + + public ManagedTopicLoadBalancer(CamelContext context, TopicLoadBalancer processor, LoadBalanceDefinition definition) { + super(context, processor, definition); + this.processor = processor; + } + + @Override + public Integer getSize() { + return processor.getProcessors().size(); + } +} http://git-wip-us.apache.org/repos/asf/camel/blob/72a06ac3/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedWeightedLoadBalancer.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedWeightedLoadBalancer.java b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedWeightedLoadBalancer.java new file mode 100644 index 0000000..1420be9 --- /dev/null +++ b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedWeightedLoadBalancer.java @@ -0,0 +1,77 @@ +/** + * 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.camel.management.mbean; + +import org.apache.camel.CamelContext; +import org.apache.camel.api.management.ManagedResource; +import org.apache.camel.api.management.mbean.ManagedWeightedBalancerMBean; +import org.apache.camel.model.LoadBalanceDefinition; +import org.apache.camel.model.loadbalancer.WeightedLoadBalancerDefinition; +import org.apache.camel.processor.loadbalancer.WeightedLoadBalancer; + +/** + * @version + */ +@ManagedResource(description = "Managed Weighted LoadBalancer") +public class ManagedWeightedLoadBalancer extends ManagedProcessor implements ManagedWeightedBalancerMBean { + private final WeightedLoadBalancer processor; + + public ManagedWeightedLoadBalancer(CamelContext context, WeightedLoadBalancer processor, LoadBalanceDefinition definition) { + super(context, processor, definition); + this.processor = processor; + } + + @Override + public Integer getSize() { + return processor.getProcessors().size(); + } + + @Override + public LoadBalanceDefinition getDefinition() { + return (LoadBalanceDefinition) super.getDefinition(); + } + + @Override + public Boolean isRoundRobin() { + WeightedLoadBalancerDefinition weighted = (WeightedLoadBalancerDefinition) getDefinition().getLoadBalancerType(); + if (weighted != null) { + return weighted.getRoundRobin(); + } else { + return null; + } + } + + @Override + public String getDistributionRatio() { + WeightedLoadBalancerDefinition weighted = (WeightedLoadBalancerDefinition) getDefinition().getLoadBalancerType(); + if (weighted != null) { + return weighted.getDistributionRatio(); + } else { + return null; + } + } + + @Override + public String getDistributionRatioDelimiter() { + WeightedLoadBalancerDefinition weighted = (WeightedLoadBalancerDefinition) getDefinition().getLoadBalancerType(); + if (weighted != null) { + return weighted.getDistributionRatioDelimiter(); + } else { + return null; + } + } +} http://git-wip-us.apache.org/repos/asf/camel/blob/72a06ac3/camel-core/src/main/java/org/apache/camel/model/loadbalancer/WeightedLoadBalancerDefinition.java ---------------------------------------------------------------------- diff --git a/camel-core/src/main/java/org/apache/camel/model/loadbalancer/WeightedLoadBalancerDefinition.java b/camel-core/src/main/java/org/apache/camel/model/loadbalancer/WeightedLoadBalancerDefinition.java index 1a314c4..c2f18cf 100644 --- a/camel-core/src/main/java/org/apache/camel/model/loadbalancer/WeightedLoadBalancerDefinition.java +++ b/camel-core/src/main/java/org/apache/camel/model/loadbalancer/WeightedLoadBalancerDefinition.java @@ -59,11 +59,7 @@ public class WeightedLoadBalancerDefinition extends LoadBalancerDefinition { List<Integer> distributionRatioList = new ArrayList<Integer>(); try { - if (distributionRatioDelimiter == null) { - distributionRatioDelimiter = ","; - } - - String[] ratios = distributionRatio.split(distributionRatioDelimiter); + String[] ratios = distributionRatio.split(getDistributionRatioDelimiter()); for (String ratio : ratios) { distributionRatioList.add(new Integer(ratio.trim())); } @@ -107,7 +103,7 @@ public class WeightedLoadBalancerDefinition extends LoadBalancerDefinition { } public String getDistributionRatioDelimiter() { - return distributionRatioDelimiter; + return distributionRatioDelimiter == null ? "," : distributionRatioDelimiter; } /** http://git-wip-us.apache.org/repos/asf/camel/blob/72a06ac3/camel-core/src/test/java/org/apache/camel/management/ManagedTopicLoadBalancerTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedTopicLoadBalancerTest.java b/camel-core/src/test/java/org/apache/camel/management/ManagedTopicLoadBalancerTest.java new file mode 100644 index 0000000..83dfc1f --- /dev/null +++ b/camel-core/src/test/java/org/apache/camel/management/ManagedTopicLoadBalancerTest.java @@ -0,0 +1,81 @@ +/** + * 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.camel.management; + +import javax.management.MBeanServer; +import javax.management.ObjectName; +import javax.management.openmbean.TabularData; + +import org.apache.camel.ServiceStatus; +import org.apache.camel.builder.RouteBuilder; + +/** + * @version + */ +public class ManagedTopicLoadBalancerTest extends ManagementTestSupport { + + public void testManageTopicLoadBalancer() throws Exception { + // JMX tests dont work well on AIX CI servers (hangs them) + if (isPlatform("aix")) { + return; + } + + // get the stats for the route + MBeanServer mbeanServer = getMBeanServer(); + + // get the object name for the delayer + ObjectName on = ObjectName.getInstance("org.apache.camel:context=camel-1,type=processors,name=\"mysend\""); + + // should be on route1 + String routeId = (String) mbeanServer.getAttribute(on, "RouteId"); + assertEquals("route1", routeId); + + String camelId = (String) mbeanServer.getAttribute(on, "CamelId"); + assertEquals("camel-1", camelId); + + String state = (String) mbeanServer.getAttribute(on, "State"); + assertEquals(ServiceStatus.Started.name(), state); + + Integer size = (Integer) mbeanServer.getAttribute(on, "Size"); + assertEquals(2, size.intValue()); + + TabularData data = (TabularData) mbeanServer.invoke(on, "explain", new Object[]{false}, new String[]{"boolean"}); + assertNotNull(data); + assertEquals(2, data.size()); + + data = (TabularData) mbeanServer.invoke(on, "explain", new Object[]{true}, new String[]{"boolean"}); + assertNotNull(data); + assertEquals(5, data.size()); + + String json = (String) mbeanServer.invoke(on, "informationJson", null, null); + assertNotNull(json); + assertTrue(json.contains("\"description\": \"Balances message processing among a number of nodes")); + } + + @Override + protected RouteBuilder createRouteBuilder() throws Exception { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + from("direct:start") + .loadBalance().topic().id("mysend") + .to("mock:foo", "mock:bar"); + } + }; + } + +} http://git-wip-us.apache.org/repos/asf/camel/blob/72a06ac3/camel-core/src/test/java/org/apache/camel/management/ManagedWeightedLoadBalancerTest.java ---------------------------------------------------------------------- diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedWeightedLoadBalancerTest.java b/camel-core/src/test/java/org/apache/camel/management/ManagedWeightedLoadBalancerTest.java new file mode 100644 index 0000000..6292f48 --- /dev/null +++ b/camel-core/src/test/java/org/apache/camel/management/ManagedWeightedLoadBalancerTest.java @@ -0,0 +1,90 @@ +/** + * 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.camel.management; + +import javax.management.MBeanServer; +import javax.management.ObjectName; +import javax.management.openmbean.TabularData; + +import org.apache.camel.ServiceStatus; +import org.apache.camel.builder.RouteBuilder; + +/** + * @version + */ +public class ManagedWeightedLoadBalancerTest extends ManagementTestSupport { + + public void testManageWeightedLoadBalancer() throws Exception { + // JMX tests dont work well on AIX CI servers (hangs them) + if (isPlatform("aix")) { + return; + } + + // get the stats for the route + MBeanServer mbeanServer = getMBeanServer(); + + // get the object name for the delayer + ObjectName on = ObjectName.getInstance("org.apache.camel:context=camel-1,type=processors,name=\"mysend\""); + + // should be on route1 + String routeId = (String) mbeanServer.getAttribute(on, "RouteId"); + assertEquals("route1", routeId); + + String camelId = (String) mbeanServer.getAttribute(on, "CamelId"); + assertEquals("camel-1", camelId); + + String state = (String) mbeanServer.getAttribute(on, "State"); + assertEquals(ServiceStatus.Started.name(), state); + + Integer size = (Integer) mbeanServer.getAttribute(on, "Size"); + assertEquals(2, size.intValue()); + + Boolean roundRobin = (Boolean) mbeanServer.getAttribute(on, "RoundRobin"); + assertEquals(true, roundRobin.booleanValue()); + + String ratio = (String) mbeanServer.getAttribute(on, "DistributionRatio"); + assertEquals("1,2", ratio); + + String delim = (String) mbeanServer.getAttribute(on, "DistributionRatioDelimiter"); + assertEquals(",", delim); + + TabularData data = (TabularData) mbeanServer.invoke(on, "explain", new Object[]{false}, new String[]{"boolean"}); + assertNotNull(data); + assertEquals(2, data.size()); + + data = (TabularData) mbeanServer.invoke(on, "explain", new Object[]{true}, new String[]{"boolean"}); + assertNotNull(data); + assertEquals(5, data.size()); + + String json = (String) mbeanServer.invoke(on, "informationJson", null, null); + assertNotNull(json); + assertTrue(json.contains("\"description\": \"Balances message processing among a number of nodes")); + } + + @Override + protected RouteBuilder createRouteBuilder() throws Exception { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + from("direct:start") + .loadBalance().weighted(true, "1,2").id("mysend") + .to("mock:foo", "mock:bar"); + } + }; + } + +}
