Updated Branches: refs/heads/javelin cdf551166 -> 98a1295fb
Add transport implementation for RPC/Async framework Project: http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/commit/98a1295f Tree: http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/tree/98a1295f Diff: http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/diff/98a1295f Branch: refs/heads/javelin Commit: 98a1295fb3565ffbe725e5e83beb3f64d519e9b1 Parents: cdf5511 Author: Kelven Yang <[email protected]> Authored: Mon Dec 10 18:10:53 2012 -0800 Committer: Kelven Yang <[email protected]> Committed: Mon Dec 10 18:11:18 2012 -0800 ---------------------------------------------------------------------- .../framework/messaging/RpcAddressable.java | 23 --- .../framework/messaging/RpcProvider.java | 8 +- .../framework/messaging/RpcProviderImpl.java | 8 +- .../framework/messaging/TransportAddress.java | 123 +++++++++++++++ .../messaging/TransportAddressMapper.java | 23 +++ .../framework/messaging/TransportDataPdu.java | 45 ++++++ .../framework/messaging/TransportEndpointSite.java | 73 +++++++++ .../framework/messaging/TransportPdu.java | 40 +++++ .../framework/messaging/TransportProvider.java | 4 +- .../messaging/client/ClientTransportProvider.java | 8 +- .../messaging/server/ServerTransportProvider.java | 107 ++++++++++++- .../AsyncSampleEventDrivenStyleCaller.java | 9 +- .../framework/messaging/SampleComponent.java | 2 +- 13 files changed, 428 insertions(+), 45 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcAddressable.java ---------------------------------------------------------------------- diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcAddressable.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcAddressable.java deleted file mode 100644 index 0bf62cf..0000000 --- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcAddressable.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 RpcAddressable { - String getAddress(); -} http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcProvider.java ---------------------------------------------------------------------- diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcProvider.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcProvider.java index 73d55b2..3b766e8 100644 --- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcProvider.java +++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcProvider.java @@ -24,16 +24,16 @@ public interface RpcProvider extends TransportMultiplexier { void setMessageSerializer(MessageSerializer messageSerializer); MessageSerializer getMessageSerializer(); - void registerRpcServiceEndpoint(String serviceAddress, RpcServiceEndpoint rpcEndpoint); - void unregisteRpcServiceEndpoint(RpcAddressable serviceAddress, RpcServiceEndpoint rpcEndpoint); + void registerRpcServiceEndpoint(RpcServiceEndpoint rpcEndpoint); + void unregisteRpcServiceEndpoint(RpcServiceEndpoint rpcEndpoint); RpcClientCall newCall(String targetAddress); - RpcClientCall newCall(RpcAddressable targetAddress); + RpcClientCall newCall(TransportAddressMapper targetAddress); // // low-level public API // - RpcClientCall newCall(TransportEndpoint sourceEndpoint, RpcAddressable targetAddress); + RpcClientCall newCall(TransportEndpoint sourceEndpoint, TransportAddressMapper targetAddress); RpcClientCall newCall(TransportEndpoint sourceEndpoint, String targetAddress); void registerCall(RpcClientCall call); http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcProviderImpl.java ---------------------------------------------------------------------- diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcProviderImpl.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcProviderImpl.java index cae85ad..d4355fa 100644 --- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcProviderImpl.java +++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/RpcProviderImpl.java @@ -69,14 +69,14 @@ public class RpcProviderImpl implements RpcProvider { } @Override - public void registerRpcServiceEndpoint(String serviceAddress, RpcServiceEndpoint rpcEndpoint) { + public void registerRpcServiceEndpoint(RpcServiceEndpoint rpcEndpoint) { synchronized(_serviceEndpoints) { _serviceEndpoints.add(rpcEndpoint); } } @Override - public void unregisteRpcServiceEndpoint(RpcAddressable serviceAddress, RpcServiceEndpoint rpcEndpoint) { + public void unregisteRpcServiceEndpoint(RpcServiceEndpoint rpcEndpoint) { synchronized(_serviceEndpoints) { _serviceEndpoints.remove(rpcEndpoint); } @@ -94,7 +94,7 @@ public class RpcProviderImpl implements RpcProvider { } @Override - public RpcClientCall newCall(TransportEndpoint sourceEndpoint, RpcAddressable targetAddress) { + public RpcClientCall newCall(TransportEndpoint sourceEndpoint, TransportAddressMapper targetAddress) { long callTag = getNextCallTag(); RpcClientCallImpl call = new RpcClientCallImpl(this); call.setSourceAddress(sourceEndpoint.getEndpointAddress()); @@ -112,7 +112,7 @@ public class RpcProviderImpl implements RpcProvider { } @Override - public RpcClientCall newCall(RpcAddressable targetAddress) { + public RpcClientCall newCall(TransportAddressMapper targetAddress) { return newCall(targetAddress.getAddress()); } http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportAddress.java ---------------------------------------------------------------------- diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportAddress.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportAddress.java new file mode 100644 index 0000000..49a4515 --- /dev/null +++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportAddress.java @@ -0,0 +1,123 @@ +/* + * 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.util.Random; + +public class TransportAddress { + public final static String LOCAL_SERVICE_NODE = ""; + + private String _nodeId = LOCAL_SERVICE_NODE; + private String _endpointId; + private int _magic; + + public TransportAddress(String nodeId, String endpointId) { + assert(nodeId != null); + assert(endpointId != null); + assert(nodeId.indexOf(".") < 0); + assert(endpointId.indexOf(".") < 0); + + _nodeId = nodeId; + _endpointId = endpointId; + _magic = new Random().nextInt(); + } + + public TransportAddress(String nodeId, String endpointId, int magic) { + assert(nodeId != null); + assert(endpointId != null); + assert(nodeId.indexOf(".") < 0); + assert(endpointId.indexOf(".") < 0); + + _nodeId = nodeId; + _endpointId = endpointId; + _magic = magic; + } + + public String getNodeId() { + return _nodeId; + } + + public TransportAddress setNodeId(String nodeId) { + _nodeId = nodeId; + return this; + } + + public String getEndpointId() { + return _endpointId; + } + + public TransportAddress setEndpointId(String endpointId) { + _endpointId = endpointId; + return this; + } + + public static TransportAddress fromAddressString(String addressString) { + if(addressString == null || addressString.isEmpty()) + return null; + + String tokens[] = addressString.split("\\."); + if(tokens.length != 3) + return null; + + return new TransportAddress(tokens[0], tokens[1], Integer.parseInt(tokens[2])); + } + + public static TransportAddress getLocalPredefinedTransportAddress(String predefinedIdentifier) { + return new TransportAddress(LOCAL_SERVICE_NODE, predefinedIdentifier, 0); + } + + @Override + public int hashCode() { + int hashCode = _magic; + hashCode = (hashCode << 3) ^ _nodeId.hashCode(); + hashCode = (hashCode << 3) ^ _endpointId.hashCode(); + + return hashCode; + } + + @Override + public boolean equals(Object other) { + if(other == null) + return false; + + if(!(other instanceof TransportAddress)) + return false; + + if(this == other) + return true; + + return _nodeId.equals(((TransportAddress)other)._nodeId) && + _endpointId.equals(((TransportAddress)other)._endpointId) && + _magic == ((TransportAddress)other)._magic; + } + + @Override + public String toString() { + StringBuffer sb = new StringBuffer(); + if(_nodeId != null) + sb.append(_nodeId); + sb.append("."); + sb.append(_endpointId); + sb.append("."); + sb.append(_magic); + + return sb.toString(); + } +} http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportAddressMapper.java ---------------------------------------------------------------------- diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportAddressMapper.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportAddressMapper.java new file mode 100644 index 0000000..11b4366 --- /dev/null +++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportAddressMapper.java @@ -0,0 +1,23 @@ +/* + * 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 TransportAddressMapper { + String getAddress(); +} http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportDataPdu.java ---------------------------------------------------------------------- diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportDataPdu.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportDataPdu.java new file mode 100644 index 0000000..1e6b323 --- /dev/null +++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportDataPdu.java @@ -0,0 +1,45 @@ +/* + * 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; + +@OnwireName(name="TransportDataPdu") +public class TransportDataPdu extends TransportPdu { + + private String _multiplexier; + private String _content; + + public TransportDataPdu() { + } + + public String getMultiplexier() { + return _multiplexier; + } + + public void setMultiplexier(String multiplexier) { + _multiplexier = multiplexier; + } + + public String getContent() { + return _content; + } + + public void setContent(String content) { + _content = content; + } +} http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportEndpointSite.java ---------------------------------------------------------------------- diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportEndpointSite.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportEndpointSite.java new file mode 100644 index 0000000..4af7772 --- /dev/null +++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportEndpointSite.java @@ -0,0 +1,73 @@ +/* + * 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.util.ArrayList; +import java.util.List; + +public class TransportEndpointSite { + private TransportEndpoint _endpoint; + private TransportAddress _address; + + private List<TransportPdu> _outputQueue = new ArrayList<TransportPdu>(); + + public TransportEndpointSite(TransportEndpoint endpoint, TransportAddress address) { + assert(endpoint != null); + assert(address != null); + + _endpoint = endpoint; + _address = address; + } + + public TransportEndpoint getEndpoint() { + return _endpoint; + } + + public TransportAddress getAddress() { + return _address; + } + + public void setAddress(TransportAddress address) { + _address = address; + } + + public void addOutputPdu(TransportPdu pdu) { + synchronized(this) { + _outputQueue.add(pdu); + } + + processOutput(); + } + + public TransportPdu getNextOutputPdu() { + synchronized(this) { + if(_outputQueue.size() > 0) + return _outputQueue.remove(0); + } + + return null; + } + + private void processOutput() { + TransportPdu pdu; + while((pdu = getNextOutputPdu()) != null) { + // ??? + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportPdu.java ---------------------------------------------------------------------- diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportPdu.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportPdu.java new file mode 100644 index 0000000..1bfb367 --- /dev/null +++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportPdu.java @@ -0,0 +1,40 @@ +/* + * 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 TransportPdu { + protected String _sourceAddress; + protected String _destAddress; + + public TransportPdu() { + } + + public String getSourceAddress() { return _sourceAddress; } + public void setSourceAddress(String sourceAddress) { + _sourceAddress = sourceAddress; + } + + public String getDestAddress() { + return _destAddress; + } + + public void setDestAddress(String destAddress) { + _destAddress = destAddress; + } +} http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportProvider.java ---------------------------------------------------------------------- diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportProvider.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportProvider.java index c843b06..132aa9a 100644 --- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportProvider.java +++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/TransportProvider.java @@ -19,8 +19,8 @@ package org.apache.cloudstack.framework.messaging; public interface TransportProvider { - void attach(TransportEndpoint endpoint, String predefinedAddress); - void detach(TransportEndpoint endpoint); + boolean attach(TransportEndpoint endpoint, String predefinedAddress); + boolean detach(TransportEndpoint endpoint); void sendMessage(String soureEndpointAddress, String targetEndpointAddress, String multiplexier, String message); http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientTransportProvider.java ---------------------------------------------------------------------- diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientTransportProvider.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientTransportProvider.java index 551838e..2189c7a 100644 --- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientTransportProvider.java +++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/client/ClientTransportProvider.java @@ -24,13 +24,16 @@ import org.apache.cloudstack.framework.messaging.TransportProvider; public class ClientTransportProvider implements TransportProvider { @Override - public void attach(TransportEndpoint endpoint, String predefinedAddress) { + public boolean attach(TransportEndpoint endpoint, String predefinedAddress) { // TODO Auto-generated method stub + return false; } @Override - public void detach(TransportEndpoint endpoint) { + public boolean detach(TransportEndpoint endpoint) { // TODO Auto-generated method stub + + return false; } @Override @@ -38,5 +41,4 @@ public class ClientTransportProvider implements TransportProvider { String multiplexier, String message) { // TODO } - } http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/src/org/apache/cloudstack/framework/messaging/server/ServerTransportProvider.java ---------------------------------------------------------------------- diff --git a/framework/ipc/src/org/apache/cloudstack/framework/messaging/server/ServerTransportProvider.java b/framework/ipc/src/org/apache/cloudstack/framework/messaging/server/ServerTransportProvider.java index 332e788..02674cb 100644 --- a/framework/ipc/src/org/apache/cloudstack/framework/messaging/server/ServerTransportProvider.java +++ b/framework/ipc/src/org/apache/cloudstack/framework/messaging/server/ServerTransportProvider.java @@ -18,26 +18,121 @@ */ package org.apache.cloudstack.framework.messaging.server; +import java.util.HashMap; +import java.util.Map; +import java.util.Random; + +import org.apache.cloudstack.framework.messaging.TransportAddress; +import org.apache.cloudstack.framework.messaging.TransportDataPdu; import org.apache.cloudstack.framework.messaging.TransportEndpoint; +import org.apache.cloudstack.framework.messaging.TransportEndpointSite; +import org.apache.cloudstack.framework.messaging.TransportPdu; import org.apache.cloudstack.framework.messaging.TransportProvider; public class ServerTransportProvider implements TransportProvider { + private String _nodeId; + private Map<String, TransportEndpointSite> _endpointMap = new HashMap<String, TransportEndpointSite>(); + + private int _nextEndpointId = new Random().nextInt(); + + public ServerTransportProvider() { + } + + public String getNodeId() { return _nodeId; } + public void setNodeId(String nodeId) { + _nodeId = nodeId; + } + @Override - public void attach(TransportEndpoint endpoint, String predefinedAddress) { - // TODO Auto-generated method stub + public boolean attach(TransportEndpoint endpoint, String predefinedAddress) { + + TransportAddress transportAddress; + String endpointId; + if(predefinedAddress != null && !predefinedAddress.isEmpty()) { + endpointId = predefinedAddress; + transportAddress = new TransportAddress(_nodeId, endpointId, 0); + } else { + endpointId = String.valueOf(getNextEndpointId()); + transportAddress = new TransportAddress(_nodeId, endpointId); + } + + TransportEndpointSite endpointSite; + synchronized(this) { + endpointSite = _endpointMap.get(endpointId); + if(endpointSite != null) { + // already attached + return false; + } + endpointSite = new TransportEndpointSite(endpoint, transportAddress); + _endpointMap.put(endpointId, endpointSite); + } + endpoint.onAttachConfirm(true, transportAddress.toString()); + return true; } @Override - public void detach(TransportEndpoint endpoint) { - // TODO Auto-generated method stub + public boolean detach(TransportEndpoint endpoint) { + TransportAddress transportAddress = TransportAddress.fromAddressString(endpoint.getEndpointAddress()); + if(transportAddress == null) + return false; + boolean found = false; + synchronized(this) { + TransportEndpointSite endpointSite = _endpointMap.get(transportAddress.getEndpointId()); + if(endpointSite.getAddress().equals(transportAddress)) { + found = true; + _endpointMap.remove(transportAddress.getEndpointId()); + } + } + + if(found) { + endpoint.onDetachIndication(endpoint.getEndpointAddress()); + return true; + } + + return false; } @Override - public void sendMessage(String soureEndpointAddress, String targetEndpointAddress, + public void sendMessage(String sourceEndpointAddress, String targetEndpointAddress, String multiplexier, String message) { - // TODO + + TransportDataPdu pdu = new TransportDataPdu(); + pdu.setSourceAddress(sourceEndpointAddress); + pdu.setDestAddress(targetEndpointAddress); + pdu.setMultiplexier(multiplexier); + pdu.setContent(message); + + dispatchPdu(pdu); + } + + private void dispatchPdu(TransportPdu pdu) { + + TransportAddress transportAddress = TransportAddress.fromAddressString(pdu.getDestAddress()); + + if(isLocalAddress(transportAddress)) { + TransportEndpointSite endpointSite = null; + synchronized(this) { + endpointSite = _endpointMap.get(transportAddress.getEndpointId()); + } + + if(endpointSite != null) + endpointSite.addOutputPdu(pdu); + } else { + // do cross-node forwarding + } + } + + private boolean isLocalAddress(TransportAddress address) { + if(address.getNodeId().equals(_nodeId) || address.getNodeId().equals(TransportAddress.LOCAL_SERVICE_NODE)) + return true; + + return false; + } + + private synchronized int getNextEndpointId() { + return _nextEndpointId++; } } http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/test/org/apache/cloudstack/framework/messaging/AsyncSampleEventDrivenStyleCaller.java ---------------------------------------------------------------------- diff --git a/framework/ipc/test/org/apache/cloudstack/framework/messaging/AsyncSampleEventDrivenStyleCaller.java b/framework/ipc/test/org/apache/cloudstack/framework/messaging/AsyncSampleEventDrivenStyleCaller.java index 849f28c..c0818db 100644 --- a/framework/ipc/test/org/apache/cloudstack/framework/messaging/AsyncSampleEventDrivenStyleCaller.java +++ b/framework/ipc/test/org/apache/cloudstack/framework/messaging/AsyncSampleEventDrivenStyleCaller.java @@ -19,7 +19,7 @@ package org.apache.cloudstack.framework.messaging; public class AsyncSampleEventDrivenStyleCaller { - AsyncSampleCallee _ds; + AsyncSampleCallee _ds = new AsyncSampleCallee(); AsyncCallbackDriver _callbackDriver; public void MethodThatWillCallAsyncMethod() { @@ -28,7 +28,7 @@ public class AsyncSampleEventDrivenStyleCaller { new AsyncCallbackDispatcher(this) .setOperationName("volume.create") .setContextParam("origVolume", vol) - .attachDriver(_callbackDriver)); + ); } @AsyncCallbackHandler(operationName="volume.create") @@ -37,4 +37,9 @@ public class AsyncSampleEventDrivenStyleCaller { TestVolume resultVol = callback.getResult(); } + + public static void main(String[] args) { + AsyncSampleEventDrivenStyleCaller caller = new AsyncSampleEventDrivenStyleCaller(); + caller.MethodThatWillCallAsyncMethod(); + } } http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/98a1295f/framework/ipc/test/org/apache/cloudstack/framework/messaging/SampleComponent.java ---------------------------------------------------------------------- diff --git a/framework/ipc/test/org/apache/cloudstack/framework/messaging/SampleComponent.java b/framework/ipc/test/org/apache/cloudstack/framework/messaging/SampleComponent.java index db1f749..25d3452 100644 --- a/framework/ipc/test/org/apache/cloudstack/framework/messaging/SampleComponent.java +++ b/framework/ipc/test/org/apache/cloudstack/framework/messaging/SampleComponent.java @@ -28,7 +28,7 @@ public class SampleComponent { public void init() { - _rpcProvider.registerRpcServiceEndpoint("AgentManager", + _rpcProvider.registerRpcServiceEndpoint( RpcServiceDispatcher.getDispatcher(this)); // subscribe to all network events (for example)
