diff options
Diffstat (limited to 'sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang')
7 files changed, 188 insertions, 58 deletions
diff --git a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangBindingProviderFactory.java b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangBindingProviderFactory.java index f4114132c1..d01d266332 100644 --- a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangBindingProviderFactory.java +++ b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangBindingProviderFactory.java @@ -21,6 +21,8 @@ package org.apache.tuscany.sca.binding.erlang.impl; import java.util.HashMap;
import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
import org.apache.tuscany.sca.binding.erlang.ErlangBinding;
import org.apache.tuscany.sca.core.ExtensionPointRegistry;
@@ -38,6 +40,8 @@ public class ErlangBindingProviderFactory implements BindingProviderFactory<ErlangBinding> {
private Map<String, ErlangNode> nodes = new HashMap<String, ErlangNode>();
+ private static final Logger logger = Logger
+ .getLogger(ErlangBindingProviderFactory.class.getName());
public ErlangBindingProviderFactory(ExtensionPointRegistry registry) {
@@ -67,8 +71,7 @@ public class ErlangBindingProviderFactory implements provider = new ErlangServiceBindingProvider(getErlangNode(binding
.getNode()), binding, service);
} catch (Exception e) {
- // TODO: log, throw, do something with this error
- e.printStackTrace();
+ logger.log(Level.WARNING, "Exception during creating ServiceBindingProvider", e);
}
return provider;
}
diff --git a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangInvoker.java b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangInvoker.java index 9ed3713db4..9a540fd5bb 100644 --- a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangInvoker.java +++ b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangInvoker.java @@ -19,12 +19,16 @@ package org.apache.tuscany.sca.binding.erlang.impl;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
import org.apache.tuscany.sca.binding.erlang.ErlangBinding;
import org.apache.tuscany.sca.binding.erlang.impl.exceptions.ErlangException;
import org.apache.tuscany.sca.binding.erlang.impl.types.TypeHelpersProxy;
import org.apache.tuscany.sca.invocation.Invoker;
import org.apache.tuscany.sca.invocation.Message;
+import com.ericsson.otp.erlang.OtpAuthException;
import com.ericsson.otp.erlang.OtpConnection;
import com.ericsson.otp.erlang.OtpEpmd;
import com.ericsson.otp.erlang.OtpErlangList;
@@ -41,32 +45,63 @@ import com.ericsson.otp.erlang.OtpSelf; */
public class ErlangInvoker implements Invoker {
+ private static final Logger logger = Logger.getLogger(ErlangInvoker.class
+ .getName());
+
private ErlangBinding binding;
- private OtpNode node;
public ErlangInvoker(ErlangBinding binding) {
this.binding = binding;
}
+ private void reportProblem(Message msg, Exception e) {
+ if (msg.getOperation().getFaultTypes().size() > 0) {
+ msg.setFaultBody(e);
+ } else {
+ // NOTE: don't throw exception if not declared
+ // TODO: externalize message?
+ logger
+ .log(Level.WARNING, "Problem while sending/receiving data",
+ e);
+ }
+ }
+
+ private boolean isCookieProvided() throws Exception {
+ return binding.getCookie() != null && binding.getCookie().length() > 0;
+ }
+
+ private String getClientNodeName() {
+ return "_connector_to_" + binding.getNode()
+ + System.currentTimeMillis();
+ }
+
private Message sendMessage(Message msg) {
OtpMbox tmpMbox = null;
+ OtpNode node = null;
try {
- String nodeName = "_connector_to_" + binding.getNode();
- node = new OtpNode(nodeName);
+ node = new OtpNode(getClientNodeName());
+ if (isCookieProvided()) {
+ node.setCookie(binding.getCookie());
+ }
tmpMbox = node.createMbox();
Object[] args = msg.getBody();
OtpErlangObject msgPayload = TypeHelpersProxy.toErlang(args);
tmpMbox.send(msg.getOperation().getName(), binding.getNode(),
msgPayload);
if (msg.getOperation().getOutputType() != null) {
- // TODO: add timeouts, timeout declaration method?
- OtpMsg resultMsg = tmpMbox.receiveMsg();
+ OtpMsg resultMsg = tmpMbox.receiveMsg(binding.getTimeout());
OtpErlangObject result = resultMsg.getMsg();
msg.setBody(TypeHelpersProxy.toJava(result, msg.getOperation()
.getOutputType().getPhysical()));
}
+ } catch (InterruptedException e) {
+ // TODO: externalize message?
+ ErlangException ee = new ErlangException(
+ "Timeout while receiving message reply", e);
+ msg.setBody(null);
+ reportProblem(msg, ee);
} catch (Exception e) {
- msg.setFaultBody(e);
+ reportProblem(msg, e);
} finally {
if (tmpMbox != null) {
tmpMbox.close();
@@ -84,8 +119,10 @@ public class ErlangInvoker implements Invoker { OtpPeer other = null;
OtpConnection connection = null;
try {
- String nodeName = "_connector_to_" + binding.getNode();
- self = new OtpSelf(nodeName);
+ self = new OtpSelf(getClientNodeName());
+ if (isCookieProvided()) {
+ self.setCookie(binding.getCookie());
+ }
other = new OtpPeer(binding.getNode());
connection = self.connect(other);
OtpErlangList params = TypeHelpersProxy
@@ -94,23 +131,18 @@ public class ErlangInvoker implements Invoker { .createRef(), binding.getModule(), msg.getOperation()
.getName(), params);
connection.send(MessageHelper.RPC_MBOX, message);
- OtpErlangObject result = connection.receiveRPC();
+ OtpErlangObject rpcResponse = connection.receive(binding
+ .getTimeout());
+ OtpErlangObject result = ((OtpErlangTuple) rpcResponse)
+ .elementAt(1);
if (MessageHelper.isfunctionUndefMessage(result)) {
// TODO: externalize message?
- Exception e = new ErlangException(
- "No such function in referenced Erlang node.");
- if (msg.getOperation().getFaultTypes().size() == 0) {
- // TODO: no way to throw exception, log it (temporary as
- // System.out)
- // TODO: do we really want not to throw any exception?
- System.out.println("PROBLEM: " + e.getMessage());
- // in this case we don't throw occured problem, so we need
- // to reset message body (if body is not cleared then
- // operation arguments would be set as operation output)
- msg.setBody(null);
- } else {
- msg.setFaultBody(e);
- }
+ Exception e = new ErlangException("No '" + binding.getModule()
+ + ":" + msg.getOperation().getName()
+ + "' operation defined on remote '" + binding.getNode()
+ + "' node.");
+ reportProblem(msg, e);
+ msg.setBody(null);
} else if (msg.getOperation().getOutputType() != null) {
if (result.getClass().equals(OtpErlangTuple.class)) {
OtpErlangObject resultBody = ((OtpErlangTuple) result)
@@ -119,8 +151,19 @@ public class ErlangInvoker implements Invoker { .getOperation().getOutputType().getPhysical()));
}
}
+ } catch (OtpAuthException e) {
+ // TODO: externalize message?
+ ErlangException ee = new ErlangException("Problem while authenticating client - check your cookie", e);
+ msg.setBody(null);
+ reportProblem(msg, ee);
+ } catch (InterruptedException e) {
+ // TODO: externalize message?
+ ErlangException ee = new ErlangException(
+ "Timeout while receiving RPC reply", e);
+ msg.setBody(null);
+ reportProblem(msg, ee);
} catch (Exception e) {
- msg.setFaultBody(e);
+ reportProblem(msg, e);
} finally {
if (connection != null) {
connection.close();
diff --git a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangNode.java b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangNode.java index 78ffdfaa49..beeec9fffb 100644 --- a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangNode.java +++ b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ErlangNode.java @@ -19,18 +19,22 @@ package org.apache.tuscany.sca.binding.erlang.impl; +import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.logging.Level; +import java.util.logging.Logger; import org.apache.tuscany.sca.binding.erlang.ErlangBinding; import org.apache.tuscany.sca.binding.erlang.impl.exceptions.ErlangException; import org.apache.tuscany.sca.interfacedef.Operation; import org.apache.tuscany.sca.runtime.RuntimeComponentService; +import com.ericsson.otp.erlang.OtpAuthException; import com.ericsson.otp.erlang.OtpConnection; import com.ericsson.otp.erlang.OtpSelf; @@ -39,6 +43,9 @@ import com.ericsson.otp.erlang.OtpSelf; */ public class ErlangNode implements Runnable { + private static final Logger logger = Logger.getLogger(ErlangNode.class + .getName()); + private Map<String, ErlangNodeElement> erlangModules = new HashMap<String, ErlangNodeElement>(); private ErlangNodeElement erlangMbox; private boolean mboxNode; @@ -53,7 +60,7 @@ public class ErlangNode implements Runnable { self = new OtpSelf(name); boolean registered = self.publishPort(); if (!registered) { - // TODO: externalize messages? + // TODO: externalize message? throw new ErlangException( "Problem with publishing service under epmd server."); } @@ -65,15 +72,20 @@ public class ErlangNode implements Runnable { } public void run() { + // FIXME: add configurable thread pools executors = Executors.newFixedThreadPool(10); while (!stopRequested) { try { OtpConnection connection = self.accept(); executors.execute(new ServiceExecutor(connection, - groupedOperations, erlangModules, erlangMbox)); - } catch (Exception e) { - // TODO: handle exception - e.printStackTrace(); + groupedOperations, erlangModules, erlangMbox, name)); + } catch (IOException e) { + // TODO: externalzie message? + logger.log(Level.WARNING, + "Error occured while accepting connection on '" + name + + "' node"); + } catch (OtpAuthException e) { + // TODO: log bad authentication attempt } } } @@ -82,8 +94,9 @@ public class ErlangNode implements Runnable { RuntimeComponentService service) throws ErlangException { if (binding.isMbox()) { if (mboxNode) { - // TODO: externalize message - // TODO: really want to throw exception? Log only? + // TODO: externalize message? + // NOTE: if mbox registered more than once for node then + // exception will be thrown throw new ErlangException("Node " + binding.getNode() + " already defined as mbox node"); } else { @@ -107,14 +120,15 @@ public class ErlangNode implements Runnable { } } else { if (erlangModules.containsKey(binding.getModule())) { - // TODO: externalize message - // TODO: really want to throw exception? Log only? + // TODO: externalize message? + // NOTE: if the same module was registered more than once than + // exception will be thrown throw new ErlangException("Module " + binding.getModule() + " already defined under " + name + " node. Duplicate module won't be started"); } else { if (erlangModules.size() == 0) { - // TODO: should ErlangNode manage its thread? + // NOTE: Erlang node is managing it's thread by itself. Just noticing. Thread selfThread = new Thread(this); selfThread.start(); } diff --git a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ServiceExecutor.java b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ServiceExecutor.java index 63c58cb696..a03d522860 100644 --- a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ServiceExecutor.java +++ b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ServiceExecutor.java @@ -23,6 +23,8 @@ import java.io.IOException; import java.lang.reflect.InvocationTargetException; import java.util.List; import java.util.Map; +import java.util.logging.Level; +import java.util.logging.Logger; import org.apache.tuscany.sca.binding.erlang.ErlangBinding; import org.apache.tuscany.sca.binding.erlang.impl.types.TypeHelpersProxy; @@ -30,8 +32,10 @@ import org.apache.tuscany.sca.interfacedef.DataType; import org.apache.tuscany.sca.interfacedef.Operation; import org.apache.tuscany.sca.runtime.RuntimeComponentService; +import com.ericsson.otp.erlang.OtpAuthException; import com.ericsson.otp.erlang.OtpConnection; import com.ericsson.otp.erlang.OtpErlangAtom; +import com.ericsson.otp.erlang.OtpErlangExit; import com.ericsson.otp.erlang.OtpErlangList; import com.ericsson.otp.erlang.OtpErlangObject; import com.ericsson.otp.erlang.OtpErlangPid; @@ -47,21 +51,25 @@ import com.ericsson.otp.erlang.OtpNode; */ public class ServiceExecutor implements Runnable { + private static final Logger logger = Logger.getLogger(ServiceExecutor.class + .getName()); private static final long RECEIVE_TIMEOUT = 60000; private Map<String, ErlangNodeElement> erlangModules; private ErlangNodeElement erlangMbox; private OtpConnection connection; private Map<String, List<Operation>> groupedOperations; + private String name; public ServiceExecutor(OtpConnection connection, Map<String, List<Operation>> groupedOperations, Map<String, ErlangNodeElement> erlangModules, - ErlangNodeElement erlangMbox) { + ErlangNodeElement erlangMbox, String name) { this.erlangModules = erlangModules; this.connection = connection; this.groupedOperations = groupedOperations; this.erlangMbox = erlangMbox; + this.name = name; } private void sendMessage(OtpConnection connection, OtpErlangPid pid, @@ -123,7 +131,8 @@ public class ServiceExecutor implements Runnable { forClasses[i] = iTypes.get(i).getPhysical(); } try { - Object result = service.getRuntimeWire(binding).invoke( + Object result = service.getRuntimeWire(binding, + service.getInterfaceContract()).invoke( operation, TypeHelpersProxy.toJavaFromList(argsList, forClasses)); @@ -169,9 +178,7 @@ public class ServiceExecutor implements Runnable { } } } catch (Exception e) { - // TODO: distinguish and describe errors! try { - e.printStackTrace(); sendMessage(connection, senderPid, senderRef, MessageHelper.ATOM_ERROR, new OtpErlangString( "Unhandled error while processing request: " @@ -179,6 +186,9 @@ public class ServiceExecutor implements Runnable { + ", message: " + e.getMessage())); } catch (IOException e1) { // error while sending error message. Can't do anything now + logger.log(Level.WARNING, "Error during sending error message", + e); + e.printStackTrace(); } } } @@ -189,7 +199,12 @@ public class ServiceExecutor implements Runnable { List<Operation> operations = groupedOperations.get(msg .getRecipientName()); if (operations == null) { - // TODO: no such mbox, send error message? + // TODO: externalize message? + // NOTE: I assume in Erlang sender doesn't get confirmation so + // message will be send + logger.log(Level.WARNING, "Node '" + name + + "' received message addressed to non exising mbox: " + + msg.getRecipientName()); } else { for (Operation operation : operations) { List<DataType> iTypes = operation.getInputType().getLogical(); @@ -203,8 +218,8 @@ public class ServiceExecutor implements Runnable { matchedOperation = operation; break; } catch (Exception e) { - // TODO Auto-generated catch block - e.printStackTrace(); + // this exception is expected while processing operation + // version with mismatched arguments } } if (matchedOperation != null) { @@ -228,22 +243,25 @@ public class ServiceExecutor implements Runnable { mbox.send(msg.getSenderPid(), response); } } catch (InvocationTargetException e) { - // TODO send some error? + // FIXME: use linking feature? send some error? e.printStackTrace(); } catch (IOException e) { - // TODO Auto-generated catch block + // FIXME: log this problem? use linking feature? send error? e.printStackTrace(); } } else { - // TODO: send some error - no mapping for such arguments - System.out - .println("TODO: send some error - no mapping for such arguments"); + // TODO: externalize message? + // NOTE: don't send error message if mapping not found + logger.log(Level.WARNING, "No mapping for such arguments in '" + + msg.getRecipientName() + "' operation in '" + name + + "' node."); } } } public void run() { try { + // TODO: should receive timeout be configured in .composite file? OtpMsg msg = connection.receiveMsg(RECEIVE_TIMEOUT); if (msg.getRecipientName().equals(MessageHelper.RPC_MBOX)) { handleRpc(msg); @@ -252,8 +270,16 @@ public class ServiceExecutor implements Runnable { } else { // message receive timeout } - } catch (Exception e) { - // TODO: log, send error? + } catch (IOException e) { + // TODO: externalize message? + logger.log(Level.WARNING, "Problem while receiving message", e); + } catch (OtpErlangExit e) { + // TODO: linking? + e.printStackTrace(); + } catch (OtpAuthException e) { + // TODO: cookies? + } catch (InterruptedException e) { + // TODO: when it could happen? e.printStackTrace(); } finally { connection.close(); diff --git a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/types/BinaryTypeHelper.java b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/types/BinaryTypeHelper.java new file mode 100644 index 0000000000..7385fe64e4 --- /dev/null +++ b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/types/BinaryTypeHelper.java @@ -0,0 +1,39 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.tuscany.sca.binding.erlang.impl.types; + +import com.ericsson.otp.erlang.OtpErlangBinary; +import com.ericsson.otp.erlang.OtpErlangObject; + +/** + * @version $Rev$ $Date$ + */ +public class BinaryTypeHelper implements TypeHelper { + + public OtpErlangObject toErlang(Object object) { + return new OtpErlangBinary((byte[])object); + } + + public Object toJava(OtpErlangObject object, Class<?> forClass) + throws Exception { + return ((OtpErlangBinary)object).binaryValue(); + } + +} diff --git a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/types/TupleTypeHelper.java b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/types/TupleTypeHelper.java index 3b5b30ac4d..324cd736ab 100644 --- a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/types/TupleTypeHelper.java +++ b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/types/TupleTypeHelper.java @@ -23,6 +23,8 @@ import java.lang.reflect.Field; import java.util.ArrayList; import java.util.List; +import org.apache.tuscany.sca.binding.erlang.impl.exceptions.ErlangException; + import com.ericsson.otp.erlang.OtpErlangObject; import com.ericsson.otp.erlang.OtpErlangTuple; @@ -42,7 +44,8 @@ public class TupleTypeHelper implements TypeHelper { OtpErlangObject member = TypeHelpersProxy.toErlang(args); tupleMembers.add(member); } catch (Exception e) { - // TODO Auto-generated catch block + // TODO: declaring toErlang method with Exception and throwing + // this? e.printStackTrace(); } } @@ -57,8 +60,12 @@ public class TupleTypeHelper implements TypeHelper { OtpErlangTuple tuple = (OtpErlangTuple) object; Field[] fields = forClass.getFields(); if (fields.length != tuple.arity()) { - // TODO: received tuple with different element count - wrong - // message, exception! + throw new ErlangException( + "Received tuple with different element count (" + + tuple.arity() + ") than expected (" + + fields.length + ")"); + // FIXME: JUnit this - received tuple with different element count - + // wrong message, exception! } result = forClass.newInstance(); for (int i = 0; i < tuple.arity(); i++) { diff --git a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/types/TypeHelpersProxy.java b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/types/TypeHelpersProxy.java index 0d0dd25a41..cbfd93796f 100644 --- a/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/types/TypeHelpersProxy.java +++ b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/types/TypeHelpersProxy.java @@ -59,15 +59,13 @@ public class TypeHelpersProxy { primitiveTypes.put(Long.class, primitiveTypes.get(long.class));
primitiveTypes.put(Float.class, primitiveTypes.get(float.class));
primitiveTypes.put(Double.class, primitiveTypes.get(double.class));
- primitiveTypes.put(String.class, primitiveTypes.get(String.class));
+ primitiveTypes.put(byte[].class, new BinaryTypeHelper());
}
private static TypeHelper getTypeHelper(Class<?> forClass) {
- TypeHelper typeHelper = null;
- if (forClass.isArray()) {
+ TypeHelper typeHelper = primitiveTypes.get(forClass);
+ if (typeHelper == null && forClass.isArray()) {
typeHelper = new ListTypeHelper();
- } else {
- typeHelper = primitiveTypes.get(forClass);
}
if (typeHelper == null) {
typeHelper = new TupleTypeHelper();
|