You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by ke...@apache.org on 2012/11/15 04:29:11 UTC
git commit: Connnect event bus to messaging layer
Updated Branches:
refs/heads/javelin cad75e722 -> 2cae51146
Connnect event bus to messaging layer
Project: http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/commit/2cae5114
Tree: http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/tree/2cae5114
Diff: http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/diff/2cae5114
Branch: refs/heads/javelin
Commit: 2cae5114644c58dab1626bfb18d6e85aeb7829f0
Parents: cad75e7
Author: Kelven Yang <ke...@gmail.com>
Authored: Wed Nov 14 19:28:34 2012 -0800
Committer: Kelven Yang <ke...@gmail.com>
Committed: Wed Nov 14 19:28:59 2012 -0800
----------------------------------------------------------------------
.../framework/container/ServerContainer.java | 27 --
.../apache/cloudstack/framework/ipc/IpcParam.java | 22 --
.../apache/cloudstack/framework/ipc/Publisher.java | 42 --
.../cloudstack/framework/ipc/ResultHandler.java | 37 --
.../apache/cloudstack/framework/ipc/Signal.java | 35 --
.../framework/ipc/SignalIntercepter.java | 23 --
.../cloudstack/framework/ipc/Subscriber.java | 39 --
.../framework/messaging/ComponentEndpoint.java | 26 +-
.../cloudstack/framework/messaging/EventBus.java | 25 ++
.../framework/messaging/EventBusBase.java | 292 ++++++++++++++
.../framework/messaging/EventHandler.java | 12 +
.../cloudstack/framework/messaging/Message.java | 29 --
.../framework/messaging/MessageHandler.java | 30 --
.../framework/messaging/MessageSystem.java | 47 ---
.../messaging/MessagingDeliveryListener.java | 24 --
.../messaging/MessagingDeliveryStrategy.java | 24 --
.../framework/messaging/MessagingProvider.java | 32 --
.../framework/messaging/MessagingSubscriber.java | 23 --
.../framework/messaging/PublishScope.java | 42 ++
.../cloudstack/framework/messaging/RpcCall.java | 4 +
.../framework/messaging/RpcCallHandler.java | 1 +
.../framework/messaging/RpcEndpoint.java | 7 +-
.../framework/messaging/RpcException.java | 17 +
.../framework/messaging/RpcIOException.java | 18 +
.../framework/messaging/RpcTimeoutException.java | 14 +
.../cloudstack/framework/messaging/Subscriber.java | 22 ++
.../framework/messaging/SubscriberDelegate.java | 27 --
.../framework/messaging/TransportEndpoint.java | 1 +
.../framework/messaging/client/ClientEventBus.java | 32 ++
.../messaging/client/ClientTransportEndpoint.java | 4 +
.../hornetq/EmbeddedMessagingProvider.java | 58 ---
.../framework/messaging/server/ServerEventBus.java | 31 ++
server/src/com/cloud/cluster/ClusterEventBus.java | 4 +-
utils/src/com/cloud/utils/events/EventBus.java | 27 --
utils/src/com/cloud/utils/events/EventBusBase.java | 293 ---------------
utils/src/com/cloud/utils/events/PublishScope.java | 23 --
utils/src/com/cloud/utils/events/Subscriber.java | 24 --
37 files changed, 539 insertions(+), 899 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/container/ServerContainer.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/container/ServerContainer.java b/framework/ipc/src/org/apache/cloudstack/framework/container/ServerContainer.java
deleted file mode 100755
index 04ec031..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/container/ServerContainer.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * 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.cloudstack.framework.container;
-
-/**
- * Container of different processes so it doesn't have to deal with message
- * to Java call.
- */
-public interface ServerContainer {
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/ipc/IpcParam.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/ipc/IpcParam.java b/framework/ipc/src/org/apache/cloudstack/framework/ipc/IpcParam.java
deleted file mode 100755
index 9bb811d..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/ipc/IpcParam.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * 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.cloudstack.framework.ipc;
-
-public @interface IpcParam {
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/ipc/Publisher.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/ipc/Publisher.java b/framework/ipc/src/org/apache/cloudstack/framework/ipc/Publisher.java
deleted file mode 100755
index 4afabd4..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/ipc/Publisher.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.cloudstack.framework.ipc;
-
-import java.util.Map;
-
-/**
- * Publish the Event
- *
- */
-public interface Publisher {
- /**
- * Publish a topic
- *
- * @param topic topic being published
- * @param content content published
- * @return true if the topic has been picked up; false if not.
- */
- boolean publish(String topic, Map<String, Object> content);
-
- /**
- * @return the name of this publisher
- */
- String getName();
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/ipc/ResultHandler.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/ipc/ResultHandler.java b/framework/ipc/src/org/apache/cloudstack/framework/ipc/ResultHandler.java
deleted file mode 100755
index 5c5c1c8..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/ipc/ResultHandler.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.cloudstack.framework.ipc;
-
-/**
- * Handles results of the ipc call
- *
- */
-public interface ResultHandler {
-
- /**
- * Signals a successful result
- */
- void signalSuccess();
-
- /**
- * Signals an error result
- * @param e exception thrown
- */
- void signalError(Exception e);
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/ipc/Signal.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/ipc/Signal.java b/framework/ipc/src/org/apache/cloudstack/framework/ipc/Signal.java
deleted file mode 100755
index 1011eaa..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/ipc/Signal.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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.cloudstack.framework.ipc;
-
-import java.lang.annotation.ElementType;
-import java.lang.annotation.Retention;
-import java.lang.annotation.RetentionPolicy;
-import java.lang.annotation.Target;
-
-/**
- * Marks a method as an ipc mechanism
- */
-@Retention(RetentionPolicy.RUNTIME)
-@Target(ElementType.METHOD)
-public @interface Signal {
- String topic();
-
- String responseTopic();
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/ipc/SignalIntercepter.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/ipc/SignalIntercepter.java b/framework/ipc/src/org/apache/cloudstack/framework/ipc/SignalIntercepter.java
deleted file mode 100755
index e0e739c..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/ipc/SignalIntercepter.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * 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.cloudstack.framework.ipc;
-
-public class SignalIntercepter {
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/ipc/Subscriber.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/ipc/Subscriber.java b/framework/ipc/src/org/apache/cloudstack/framework/ipc/Subscriber.java
deleted file mode 100755
index 0856cdd..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/ipc/Subscriber.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.cloudstack.framework.ipc;
-
-import org.apache.cloudstack.framework.messaging.Message;
-
-/**
- * Event subscriber interface
- *
- */
-public interface Subscriber {
-
- /**
- * Message received
- */
- Message receive();
-
- /**
- * @return the name of the subscriber
- */
- String getName();
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/ComponentEndpoint.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/ComponentEndpoint.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/ComponentEndpoint.java
index 0c55b30..ed9ea0b 100644
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/ComponentEndpoint.java
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/ComponentEndpoint.java
@@ -18,7 +18,7 @@
*/
package org.apache.cloudstack.framework.messaging;
-public class ComponentEndpoint implements RpcEndpoint, TransportMultiplexier {
+public class ComponentEndpoint implements RpcEndpoint, TransportMultiplexier, Subscriber {
private TransportEndpoint transportEndpoint;
private RpcProvider rpcProvider;
@@ -57,19 +57,29 @@ public class ComponentEndpoint implements RpcEndpoint, TransportMultiplexier {
}
@Override
- public String call(String targetAddress, String rpcMessage) {
- // TODO Auto-generated method stub
+ public String call(String targetAddress, String rpcMessage)
+ {
return null;
}
-
+
@Override
- public void asyncCall(String targetAddress, String rpcMessage) {
- // TODO Auto-generated method stub
+ public RpcCall asyncCall(String targetAddress, String rpcMessage) {
+ return null;
}
@Override
- public void onCall(RpcCall call) {
+ public void onCallReceive(RpcCall call) {
// TODO Auto-generated method stub
-
+ // implement annotation based call dispatching
+ }
+
+ @Override
+ public void onCallReturn(RpcCall call, String rpcReturnMessage, RpcException e) {
+ // ???
+ }
+
+ @Override
+ public void onPublishEvent(String subject, String senderAddress, String args) {
+ // TODO
}
}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventBus.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventBus.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventBus.java
new file mode 100644
index 0000000..e11a009
--- /dev/null
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventBus.java
@@ -0,0 +1,25 @@
+// 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
+// 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.cloudstack.framework.messaging;
+
+public interface EventBus {
+ void subscribe(String subject, Subscriber subscriber);
+ void unsubscribe(String subject, Subscriber subscriber);
+
+ void publish(String subject, PublishScope scope, String senderAddress, String args);
+}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventBusBase.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventBusBase.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventBusBase.java
new file mode 100644
index 0000000..fbcf648
--- /dev/null
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventBusBase.java
@@ -0,0 +1,292 @@
+// 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
+// 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.cloudstack.framework.messaging;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+
+public class EventBusBase implements EventBus {
+
+ private Gate _gate;
+ private List<ActionRecord> _pendingActions;
+
+ private SubscriptionNode _subscriberRoot;
+
+ public EventBusBase() {
+ _gate = new Gate();
+ _pendingActions = new ArrayList<ActionRecord>();
+
+ _subscriberRoot = new SubscriptionNode("/", null);
+ }
+
+ @Override
+ public void subscribe(String subject, Subscriber subscriber) {
+ assert(subject != null);
+ assert(subscriber != null);
+ if(_gate.enter()) {
+ SubscriptionNode current = locate(subject, null, true);
+ assert(current != null);
+ current.addSubscriber(subscriber);
+ _gate.leave();
+ } else {
+ synchronized(_pendingActions) {
+ _pendingActions.add(new ActionRecord(ActionType.Subscribe, subject, subscriber));
+ }
+ }
+ }
+
+ @Override
+ public void unsubscribe(String subject, Subscriber subscriber) {
+ if(_gate.enter()) {
+ SubscriptionNode current = locate(subject, null, false);
+ if(current != null)
+ current.removeSubscriber(subscriber);
+
+ _gate.leave();
+ } else {
+ synchronized(_pendingActions) {
+ _pendingActions.add(new ActionRecord(ActionType.Unsubscribe, subject, subscriber));
+ }
+ }
+ }
+
+ @Override
+ public void publish(String subject, PublishScope scope, String senderAddress,
+ String args) {
+
+ if(_gate.enter(true)) {
+
+ List<SubscriptionNode> chainFromTop = new ArrayList<SubscriptionNode>();
+ SubscriptionNode current = locate(subject, chainFromTop, false);
+
+ if(current != null)
+ current.notifySubscribers(subject, senderAddress, args);
+
+ Collections.reverse(chainFromTop);
+ for(SubscriptionNode node : chainFromTop)
+ node.notifySubscribers(subject, senderAddress, args);
+
+ _gate.leave();
+ }
+ }
+
+ private void onGateOpen() {
+ synchronized(_pendingActions) {
+ ActionRecord record = null;
+ if(_pendingActions.size() > 0) {
+ while((record = _pendingActions.remove(0)) != null) {
+ switch(record.getType()) {
+ case Subscribe :
+ {
+ SubscriptionNode current = locate(record.getSubject(), null, true);
+ assert(current != null);
+ current.addSubscriber(record.getSubscriber());
+ }
+ break;
+
+ case Unsubscribe :
+ {
+ SubscriptionNode current = locate(record.getSubject(), null, false);
+ if(current != null)
+ current.removeSubscriber(record.getSubscriber());
+ }
+ break;
+
+ default :
+ assert(false);
+ break;
+
+ }
+ }
+ }
+ }
+ }
+
+
+ private SubscriptionNode locate(String subject, List<SubscriptionNode> chainFromTop,
+ boolean createPath) {
+
+ assert(subject != null);
+
+ String[] subjectPathTokens = subject.split("\\.");
+ return locate(subjectPathTokens, _subscriberRoot, chainFromTop, createPath);
+ }
+
+ private static SubscriptionNode locate(String[] subjectPathTokens,
+ SubscriptionNode current, List<SubscriptionNode> chainFromTop, boolean createPath) {
+
+ assert(current != null);
+ assert(subjectPathTokens != null);
+ assert(subjectPathTokens.length > 0);
+
+ if(chainFromTop != null)
+ chainFromTop.add(current);
+
+ SubscriptionNode next = current.getChild(subjectPathTokens[0]);
+ if(next == null) {
+ if(createPath) {
+ next = new SubscriptionNode(subjectPathTokens[0], null);
+ current.addChild(subjectPathTokens[0], next);
+ } else {
+ return null;
+ }
+ }
+
+ if(subjectPathTokens.length > 1) {
+ return locate((String[])Arrays.copyOfRange(subjectPathTokens, 1, subjectPathTokens.length),
+ next, chainFromTop, createPath);
+ } else {
+ return next;
+ }
+ }
+
+
+ //
+ // Support inner classes
+ //
+ private static enum ActionType {
+ Subscribe,
+ Unsubscribe
+ }
+
+ private static class ActionRecord {
+ private ActionType _type;
+ private String _subject;
+ private Subscriber _subscriber;
+
+ public ActionRecord(ActionType type, String subject, Subscriber subscriber) {
+ _type = type;
+ _subject = subject;
+ _subscriber = subscriber;
+ }
+
+ public ActionType getType() {
+ return _type;
+ }
+
+ public String getSubject() {
+ return _subject;
+ }
+
+ public Subscriber getSubscriber() {
+ return _subscriber;
+ }
+ }
+
+ private class Gate {
+ private int _reentranceCount;
+ private Thread _gateOwner;
+
+ public Gate() {
+ _reentranceCount = 0;
+ _gateOwner = null;
+ }
+
+ public boolean enter() {
+ return enter(false);
+ }
+
+ public boolean enter(boolean wait) {
+ while(true) {
+ synchronized(this) {
+ if(_reentranceCount == 0) {
+ assert(_gateOwner == null);
+
+ _reentranceCount++;
+ _gateOwner = Thread.currentThread();
+ return true;
+ } else {
+ if(wait) {
+ try {
+ wait();
+ } catch (InterruptedException e) {
+ }
+ } else {
+ break;
+ }
+ }
+ }
+ }
+
+ return false;
+ }
+
+ public void leave() {
+ synchronized(this) {
+ if(_reentranceCount > 0) {
+ assert(_gateOwner == Thread.currentThread());
+
+ onGateOpen();
+ _reentranceCount--;
+ assert(_reentranceCount == 0);
+ _gateOwner = null;
+
+ notifyAll();
+ }
+ }
+ }
+ }
+
+ private static class SubscriptionNode {
+ private String _nodeKey;
+ private List<Subscriber> _subscribers;
+ private Map<String, SubscriptionNode> _children;
+
+ public SubscriptionNode(String nodeKey, Subscriber subscriber) {
+ assert(nodeKey != null);
+ _nodeKey = nodeKey;
+ _subscribers = new ArrayList<Subscriber>();
+
+ if(subscriber != null)
+ _subscribers.add(subscriber);
+
+ _children = new HashMap<String, SubscriptionNode>();
+ }
+
+ public List<Subscriber> getSubscriber() {
+ return _subscribers;
+ }
+
+ public void addSubscriber(Subscriber subscriber) {
+ _subscribers.add(subscriber);
+ }
+
+ public void removeSubscriber(Subscriber subscriber) {
+ _subscribers.remove(subscriber);
+ }
+
+ public SubscriptionNode getChild(String key) {
+ return _children.get(key);
+ }
+
+ public void addChild(String key, SubscriptionNode childNode) {
+ _children.put(key, childNode);
+ }
+
+ public void notifySubscribers(String subject, String senderAddress, String args) {
+ for(Subscriber subscriber : _subscribers) {
+ subscriber.onPublishEvent(subject, senderAddress, args);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventHandler.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventHandler.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventHandler.java
new file mode 100644
index 0000000..12e6fb8
--- /dev/null
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/EventHandler.java
@@ -0,0 +1,12 @@
+package org.apache.cloudstack.framework.messaging;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.METHOD)
+public @interface EventHandler {
+ public String topic();
+}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/Message.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/Message.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/Message.java
deleted file mode 100755
index 21617d5..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/Message.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * 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.cloudstack.framework.messaging;
-
-public interface Message {
-
- String getTitle();
- String getContent();
- String getPublisherTag();
-
- long getSendTime();
- long getReceiveTime();
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessageHandler.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessageHandler.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessageHandler.java
deleted file mode 100644
index 1061042..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessageHandler.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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.cloudstack.framework.messaging;
-
-import java.lang.annotation.ElementType;
-import java.lang.annotation.Retention;
-import java.lang.annotation.RetentionPolicy;
-import java.lang.annotation.Target;
-
-@Retention(RetentionPolicy.RUNTIME)
-@Target(ElementType.METHOD)
-public @interface MessageHandler {
- String messageTitle();
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessageSystem.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessageSystem.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessageSystem.java
deleted file mode 100755
index cc506ef..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessageSystem.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.cloudstack.framework.messaging;
-
-import org.apache.cloudstack.framework.ipc.Publisher;
-import org.apache.cloudstack.framework.ipc.Subscriber;
-
-public interface MessageSystem {
-
- /**
- * Creates the publisher
- * @param name of the publisher
- * @return publisher
- */
- Publisher createPublisher(String name);
-
- /**
- * Creates the subscriber
- * @param name of the subscriber
- * @return subscriber
- */
- Subscriber createSubscriber(String name);
-
- /**
- * registers the subscriber
- * @param subscriber subscriber
- * @param topic topic to listen to
- * @return subscriber
- */
- boolean registerSubscriber(Subscriber subscriber, String topic);
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingDeliveryListener.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingDeliveryListener.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingDeliveryListener.java
deleted file mode 100644
index be70b54..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingDeliveryListener.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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.cloudstack.framework.messaging;
-
-public interface MessagingDeliveryListener {
- void onAcknowledge(String topicChannel, Message message);
- void onTimeout(String topicChannel, Message message);
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingDeliveryStrategy.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingDeliveryStrategy.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingDeliveryStrategy.java
deleted file mode 100644
index 2a9b066..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingDeliveryStrategy.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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.cloudstack.framework.messaging;
-
-public enum MessagingDeliveryStrategy {
- atMostOneCasting,
- blindCasting
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingProvider.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingProvider.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingProvider.java
deleted file mode 100644
index 07e3d71..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingProvider.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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.cloudstack.framework.messaging;
-
-public interface MessagingProvider {
- void createChanel(String topic, MessagingDeliveryStrategy deliveryStrategy);
-
- void pulishMessage(String topic, Message message);
-
- void publishCertifiedMessage(String topic, Message message,
- int retryIntervalMillis, int timeoutMillis,
- MessagingDeliveryListener deliveryListener);
-
- void subscribe(String topicChannel, String messageTitle, MessagingSubscriber subscriber);
- void unsubscribe(String topicChannel, String messageTitle, MessagingSubscriber subscriber);
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingSubscriber.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingSubscriber.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingSubscriber.java
deleted file mode 100644
index 5019a43..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/MessagingSubscriber.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * 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.cloudstack.framework.messaging;
-
-public interface MessagingSubscriber {
- void onMessage(String topic, Message message);
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/PublishScope.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/PublishScope.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/PublishScope.java
new file mode 100644
index 0000000..fbce919
--- /dev/null
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/PublishScope.java
@@ -0,0 +1,42 @@
+// 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
+// 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.cloudstack.framework.messaging;
+
+public class PublishScope {
+ public enum Type { SINGLE, LOCAL, GLOBAL };
+
+ Type scope;
+ String address;
+
+ public PublishScope(Type scope) {
+ this.scope = scope;
+ }
+
+ public PublishScope(String address) {
+ scope = Type.SINGLE;
+ this.address = address;
+ }
+
+ public Type getType() {
+ return scope;
+ }
+
+ public String getAddress() {
+ return this.address;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcCall.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcCall.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcCall.java
index be16bdd..570b82a 100644
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcCall.java
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcCall.java
@@ -22,6 +22,10 @@ public interface RpcCall {
String getCommand();
String getContent();
String getRequestTag();
+
+ // for sender to cancel
+ void cancel();
+ // for receiver to response call
void completeCall(String rpcMessage);
}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcCallHandler.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcCallHandler.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcCallHandler.java
index 2d94064..70110ee 100644
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcCallHandler.java
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcCallHandler.java
@@ -27,4 +27,5 @@ import java.lang.annotation.Target;
@Target(ElementType.METHOD)
public @interface RpcCallHandler {
String command();
+ boolean returnHandler() default false;
}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcEndpoint.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcEndpoint.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcEndpoint.java
index 62dc973..1be5890 100644
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcEndpoint.java
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcEndpoint.java
@@ -19,8 +19,9 @@
package org.apache.cloudstack.framework.messaging;
public interface RpcEndpoint {
- public String call(String targetAddress, String rpcMessage);
- public void asyncCall(String targetAddress, String rpcMessage);
+ String call(String targetAddress, String rpcMessage);
+ RpcCall asyncCall(String targetAddress, String rpcMessage);
- void onCall(RpcCall call);
+ void onCallReceive(RpcCall call);
+ void onCallReturn(RpcCall call, String rpcReturnMessage, RpcException e);
}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcException.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcException.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcException.java
new file mode 100644
index 0000000..978f1c3
--- /dev/null
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcException.java
@@ -0,0 +1,17 @@
+package org.apache.cloudstack.framework.messaging;
+
+public class RpcException extends Exception {
+ private static final long serialVersionUID = -3164514701087423787L;
+
+ public RpcException() {
+ super();
+ }
+
+ public RpcException(String message) {
+ super(message);
+ }
+
+ public RpcException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcIOException.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcIOException.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcIOException.java
new file mode 100644
index 0000000..8b09ca9
--- /dev/null
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcIOException.java
@@ -0,0 +1,18 @@
+package org.apache.cloudstack.framework.messaging;
+
+public class RpcIOException extends RpcException {
+
+ private static final long serialVersionUID = -6108039302920641533L;
+
+ public RpcIOException() {
+ super();
+ }
+
+ public RpcIOException(String message) {
+ super(message);
+ }
+
+ public RpcIOException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcTimeoutException.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcTimeoutException.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcTimeoutException.java
new file mode 100644
index 0000000..c0521c5
--- /dev/null
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcTimeoutException.java
@@ -0,0 +1,14 @@
+package org.apache.cloudstack.framework.messaging;
+
+public class RpcTimeoutException extends RpcException {
+
+ private static final long serialVersionUID = -3618654987984665833L;
+
+ public RpcTimeoutException() {
+ super();
+ }
+
+ public RpcTimeoutException(String message) {
+ super(message);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/Subscriber.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/Subscriber.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/Subscriber.java
new file mode 100644
index 0000000..3eb3fc6
--- /dev/null
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/Subscriber.java
@@ -0,0 +1,22 @@
+// 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
+// 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.cloudstack.framework.messaging;
+
+public interface Subscriber {
+ void onPublishEvent(String subject, String senderAddress, String args);
+}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/SubscriberDelegate.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/SubscriberDelegate.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/SubscriberDelegate.java
deleted file mode 100644
index 40a2eea..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/SubscriberDelegate.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/*
- * 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.cloudstack.framework.messaging;
-
-public class SubscriberDelegate implements MessagingSubscriber {
-
- public void onMessage(String topic, Message message) {
- // ???
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportEndpoint.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportEndpoint.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportEndpoint.java
index 68982e2..3a98681 100644
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportEndpoint.java
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportEndpoint.java
@@ -20,6 +20,7 @@ package org.apache.cloudstack.framework.messaging;
public interface TransportEndpoint {
void onAttachConfirm(String endpointAddress);
+ void onDetachIndication(String endpointAddress);
void registerMultiplexier(String name, TransportMultiplexier multiplexier);
void unregisterMultiplexier(String name);
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientEventBus.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientEventBus.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientEventBus.java
new file mode 100644
index 0000000..68303bc
--- /dev/null
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientEventBus.java
@@ -0,0 +1,32 @@
+/*
+ * 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.cloudstack.framework.messaging.client;
+
+import org.apache.cloudstack.framework.messaging.EventBusBase;
+import org.apache.cloudstack.framework.messaging.TransportMultiplexier;
+
+public class ClientEventBus extends EventBusBase implements TransportMultiplexier {
+
+ @Override
+ public void onTransportMessage(String senderEndpointAddress,
+ String targetEndpointAddress, String multiplexer, String message) {
+ // TODO Auto-generated method stub
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientTransportEndpoint.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientTransportEndpoint.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientTransportEndpoint.java
index 8ee4b8f..49a1eb6 100644
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientTransportEndpoint.java
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientTransportEndpoint.java
@@ -27,6 +27,10 @@ public class ClientTransportEndpoint implements TransportEndpoint {
public void onAttachConfirm(String endpointAddress) {
// TODO Auto-generated method stub
}
+
+ @Override
+ public void onDetachIndication(String endpointAddress) {
+ }
@Override
public void registerMultiplexier(String name,
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/hornetq/EmbeddedMessagingProvider.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/hornetq/EmbeddedMessagingProvider.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/hornetq/EmbeddedMessagingProvider.java
deleted file mode 100644
index d928128..0000000
--- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/hornetq/EmbeddedMessagingProvider.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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.cloudstack.framework.messaging.hornetq;
-
-import org.apache.cloudstack.framework.messaging.Message;
-import org.apache.cloudstack.framework.messaging.MessagingDeliveryListener;
-import org.apache.cloudstack.framework.messaging.MessagingDeliveryStrategy;
-import org.apache.cloudstack.framework.messaging.MessagingProvider;
-import org.apache.cloudstack.framework.messaging.MessagingSubscriber;
-
-public class EmbeddedMessagingProvider implements MessagingProvider {
-
- public void createChanel(String topic,
- MessagingDeliveryStrategy deliveryStrategy) {
- // TODO Auto-generated method stub
-
- }
-
- public void pulishMessage(String topic, Message message) {
- // TODO Auto-generated method stub
-
- }
-
- public void publishCertifiedMessage(String topic, Message message,
- int retryIntervalMillis, int timeoutMillis,
- MessagingDeliveryListener deliveryListener) {
- // TODO Auto-generated method stub
-
- }
-
- public void subscribe(String topicChannel, String messageTitle,
- MessagingSubscriber subscriber) {
- // TODO Auto-generated method stub
-
- }
-
- public void unsubscribe(String topicChannel, String messageTitle,
- MessagingSubscriber subscriber) {
- // TODO Auto-generated method stub
-
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/framework/ipc/src/org/apache/cloudstack/framework/messaging/server/ServerEventBus.java
----------------------------------------------------------------------
diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/server/ServerEventBus.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/server/ServerEventBus.java
new file mode 100644
index 0000000..ff02cb8
--- /dev/null
+++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/server/ServerEventBus.java
@@ -0,0 +1,31 @@
+/*
+ * 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.cloudstack.framework.messaging.server;
+
+import org.apache.cloudstack.framework.messaging.EventBusBase;
+import org.apache.cloudstack.framework.messaging.TransportMultiplexier;
+
+public class ServerEventBus extends EventBusBase implements TransportMultiplexier {
+
+ @Override
+ public void onTransportMessage(String senderEndpointAddress,
+ String targetEndpointAddress, String multiplexer, String message) {
+ // TODO Auto-generated method stub
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/server/src/com/cloud/cluster/ClusterEventBus.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/cluster/ClusterEventBus.java b/server/src/com/cloud/cluster/ClusterEventBus.java
index 4df5d37..63ceaea 100644
--- a/server/src/com/cloud/cluster/ClusterEventBus.java
+++ b/server/src/com/cloud/cluster/ClusterEventBus.java
@@ -4,10 +4,10 @@ import java.io.Serializable;
import javax.annotation.PostConstruct;
+import org.apache.cloudstack.framework.messaging.EventBusBase;
+import org.apache.cloudstack.framework.messaging.PublishScope;
import org.springframework.stereotype.Component;
-import com.cloud.utils.events.EventBusBase;
-import com.cloud.utils.events.PublishScope;
@Component
public class ClusterEventBus extends EventBusBase {
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/utils/src/com/cloud/utils/events/EventBus.java
----------------------------------------------------------------------
diff --git a/utils/src/com/cloud/utils/events/EventBus.java b/utils/src/com/cloud/utils/events/EventBus.java
deleted file mode 100644
index c1b6f70..0000000
--- a/utils/src/com/cloud/utils/events/EventBus.java
+++ /dev/null
@@ -1,27 +0,0 @@
-// 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
-// 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 com.cloud.utils.events;
-
-import java.io.Serializable;
-
-public interface EventBus {
- void subscribe(String subject, Subscriber subscriber);
- void unsubscribe(String subject, Subscriber subscriber);
-
- void publish(String subject, PublishScope scope, Object sender, Serializable args);
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/utils/src/com/cloud/utils/events/EventBusBase.java
----------------------------------------------------------------------
diff --git a/utils/src/com/cloud/utils/events/EventBusBase.java b/utils/src/com/cloud/utils/events/EventBusBase.java
deleted file mode 100644
index cd10f1d..0000000
--- a/utils/src/com/cloud/utils/events/EventBusBase.java
+++ /dev/null
@@ -1,293 +0,0 @@
-// 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
-// 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 com.cloud.utils.events;
-
-import java.io.Serializable;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import edu.emory.mathcs.backport.java.util.Arrays;
-import edu.emory.mathcs.backport.java.util.Collections;
-
-public class EventBusBase implements EventBus {
-
- private Gate _gate;
- private List<ActionRecord> _pendingActions;
-
- private SubscriptionNode _subscriberRoot;
-
- public EventBusBase() {
- _gate = new Gate();
- _pendingActions = new ArrayList<ActionRecord>();
-
- _subscriberRoot = new SubscriptionNode("/", null);
- }
-
- @Override
- public void subscribe(String subject, Subscriber subscriber) {
- assert(subject != null);
- assert(subscriber != null);
- if(_gate.enter()) {
- SubscriptionNode current = locate(subject, null, true);
- assert(current != null);
- current.addSubscriber(subscriber);
- _gate.leave();
- } else {
- synchronized(_pendingActions) {
- _pendingActions.add(new ActionRecord(ActionType.Subscribe, subject, subscriber));
- }
- }
- }
-
- @Override
- public void unsubscribe(String subject, Subscriber subscriber) {
- if(_gate.enter()) {
- SubscriptionNode current = locate(subject, null, false);
- if(current != null)
- current.removeSubscriber(subscriber);
-
- _gate.leave();
- } else {
- synchronized(_pendingActions) {
- _pendingActions.add(new ActionRecord(ActionType.Unsubscribe, subject, subscriber));
- }
- }
- }
-
- @Override
- public void publish(String subject, PublishScope scope, Object sender,
- Serializable args) {
-
- if(_gate.enter(true)) {
-
- List<SubscriptionNode> chainFromTop = new ArrayList<SubscriptionNode>();
- SubscriptionNode current = locate(subject, chainFromTop, false);
-
- if(current != null)
- current.notifySubscribers(subject, sender, args);
-
- Collections.reverse(chainFromTop);
- for(SubscriptionNode node : chainFromTop)
- node.notifySubscribers(subject, sender, args);
-
- _gate.leave();
- }
- }
-
- private void onGateOpen() {
- synchronized(_pendingActions) {
- ActionRecord record = null;
- if(_pendingActions.size() > 0) {
- while((record = _pendingActions.remove(0)) != null) {
- switch(record.getType()) {
- case Subscribe :
- {
- SubscriptionNode current = locate(record.getSubject(), null, true);
- assert(current != null);
- current.addSubscriber(record.getSubscriber());
- }
- break;
-
- case Unsubscribe :
- {
- SubscriptionNode current = locate(record.getSubject(), null, false);
- if(current != null)
- current.removeSubscriber(record.getSubscriber());
- }
- break;
-
- default :
- assert(false);
- break;
-
- }
- }
- }
- }
- }
-
-
- private SubscriptionNode locate(String subject, List<SubscriptionNode> chainFromTop,
- boolean createPath) {
-
- assert(subject != null);
-
- String[] subjectPathTokens = subject.split("\\.");
- return locate(subjectPathTokens, _subscriberRoot, chainFromTop, createPath);
- }
-
- private static SubscriptionNode locate(String[] subjectPathTokens,
- SubscriptionNode current, List<SubscriptionNode> chainFromTop, boolean createPath) {
-
- assert(current != null);
- assert(subjectPathTokens != null);
- assert(subjectPathTokens.length > 0);
-
- if(chainFromTop != null)
- chainFromTop.add(current);
-
- SubscriptionNode next = current.getChild(subjectPathTokens[0]);
- if(next == null) {
- if(createPath) {
- next = new SubscriptionNode(subjectPathTokens[0], null);
- current.addChild(subjectPathTokens[0], next);
- } else {
- return null;
- }
- }
-
- if(subjectPathTokens.length > 1) {
- return locate((String[])Arrays.copyOfRange(subjectPathTokens, 1, subjectPathTokens.length),
- next, chainFromTop, createPath);
- } else {
- return next;
- }
- }
-
-
- //
- // Support inner classes
- //
- private static enum ActionType {
- Subscribe,
- Unsubscribe
- }
-
- private static class ActionRecord {
- private ActionType _type;
- private String _subject;
- private Subscriber _subscriber;
-
- public ActionRecord(ActionType type, String subject, Subscriber subscriber) {
- _type = type;
- _subject = subject;
- _subscriber = subscriber;
- }
-
- public ActionType getType() {
- return _type;
- }
-
- public String getSubject() {
- return _subject;
- }
-
- public Subscriber getSubscriber() {
- return _subscriber;
- }
- }
-
- private class Gate {
- private int _reentranceCount;
- private Thread _gateOwner;
-
- public Gate() {
- _reentranceCount = 0;
- _gateOwner = null;
- }
-
- public boolean enter() {
- return enter(false);
- }
-
- public boolean enter(boolean wait) {
- while(true) {
- synchronized(this) {
- if(_reentranceCount == 0) {
- assert(_gateOwner == null);
-
- _reentranceCount++;
- _gateOwner = Thread.currentThread();
- return true;
- } else {
- if(wait) {
- try {
- wait();
- } catch (InterruptedException e) {
- }
- } else {
- break;
- }
- }
- }
- }
-
- return false;
- }
-
- public void leave() {
- synchronized(this) {
- if(_reentranceCount > 0) {
- assert(_gateOwner == Thread.currentThread());
-
- onGateOpen();
- _reentranceCount--;
- assert(_reentranceCount == 0);
- _gateOwner = null;
-
- notifyAll();
- }
- }
- }
- }
-
- private static class SubscriptionNode {
- private String _nodeKey;
- private List<Subscriber> _subscribers;
- private Map<String, SubscriptionNode> _children;
-
- public SubscriptionNode(String nodeKey, Subscriber subscriber) {
- assert(nodeKey != null);
- _nodeKey = nodeKey;
- _subscribers = new ArrayList<Subscriber>();
-
- if(subscriber != null)
- _subscribers.add(subscriber);
-
- _children = new HashMap<String, SubscriptionNode>();
- }
-
- public List<Subscriber> getSubscriber() {
- return _subscribers;
- }
-
- public void addSubscriber(Subscriber subscriber) {
- _subscribers.add(subscriber);
- }
-
- public void removeSubscriber(Subscriber subscriber) {
- _subscribers.remove(subscriber);
- }
-
- public SubscriptionNode getChild(String key) {
- return _children.get(key);
- }
-
- public void addChild(String key, SubscriptionNode childNode) {
- _children.put(key, childNode);
- }
-
- public void notifySubscribers(String subject, Object sender, Serializable args) {
- for(Subscriber subscriber : _subscribers) {
- subscriber.onPublishEvent(subject, sender, args);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/utils/src/com/cloud/utils/events/PublishScope.java
----------------------------------------------------------------------
diff --git a/utils/src/com/cloud/utils/events/PublishScope.java b/utils/src/com/cloud/utils/events/PublishScope.java
deleted file mode 100644
index bab852a..0000000
--- a/utils/src/com/cloud/utils/events/PublishScope.java
+++ /dev/null
@@ -1,23 +0,0 @@
-// 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
-// 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 com.cloud.utils.events;
-
-public enum PublishScope {
- LOCAL,
- GLOBAL
-}
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/2cae5114/utils/src/com/cloud/utils/events/Subscriber.java
----------------------------------------------------------------------
diff --git a/utils/src/com/cloud/utils/events/Subscriber.java b/utils/src/com/cloud/utils/events/Subscriber.java
deleted file mode 100644
index c3baa6f..0000000
--- a/utils/src/com/cloud/utils/events/Subscriber.java
+++ /dev/null
@@ -1,24 +0,0 @@
-// 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
-// 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 com.cloud.utils.events;
-
-import java.io.Serializable;
-
-public interface Subscriber {
- void onPublishEvent(String subject, Object sender, Serializable args);
-}