1 files changed, 378 insertions, 0 deletions
diff --git a/sca-java-1.x/tags/1.6-TUSCANY-3909/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ServiceExecutor.java b/sca-java-1.x/tags/1.6-TUSCANY-3909/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ServiceExecutor.java
new file mode 100644
index 0000000000..9e292fe3bb
--- /dev/null
+++ b/sca-java-1.x/tags/1.6-TUSCANY-3909/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ServiceExecutor.java
@@ -0,0 +1,378 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+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;
+import org.apache.tuscany.sca.interfacedef.DataType;
+import org.apache.tuscany.sca.interfacedef.Operation;
+import org.apache.tuscany.sca.interfacedef.java.JavaOperation;
+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.OtpErlangDecodeException;
+import com.ericsson.otp.erlang.OtpErlangExit;
+import com.ericsson.otp.erlang.OtpErlangList;
+import com.ericsson.otp.erlang.OtpErlangObject;
+import com.ericsson.otp.erlang.OtpErlangPid;
+import com.ericsson.otp.erlang.OtpErlangRef;
+import com.ericsson.otp.erlang.OtpErlangString;
+import com.ericsson.otp.erlang.OtpErlangTuple;
+import com.ericsson.otp.erlang.OtpMsg;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class ServiceExecutor implements Runnable {
+
+	private static final Logger logger = Logger.getLogger(ServiceExecutor.class
+			.getName());
+
+	private ErlangNodeElement nodeElement;
+	private OtpConnection connection;
+	private Map<String, List<Operation>> groupedOperations;
+	private String name;
+
+	public ServiceExecutor(OtpConnection connection,
+			Map<String, List<Operation>> groupedOperations,
+			ErlangNodeElement nodeElement, String name) {
+		this.connection = connection;
+		this.groupedOperations = groupedOperations;
+		this.nodeElement = nodeElement;
+		this.name = name;
+	}
+
+	private void sendMessage(OtpConnection connection, OtpErlangPid pid,
+			OtpErlangRef ref, OtpErlangAtom head, OtpErlangObject message)
+			throws IOException {
+		OtpErlangObject tResult = null;
+		if (head != null) {
+			tResult = new OtpErlangTuple(
+					new OtpErlangObject[] { head, message });
+		} else {
+			tResult = message;
+		}
+		OtpErlangObject msg = null;
+		msg = new OtpErlangTuple(new OtpErlangObject[] { ref, tResult });
+		connection.send(pid, msg);
+	}
+
+	private void handleRpc(OtpMsg msg) {
+		OtpErlangTuple request = null;
+		OtpErlangPid senderPid = null;
+		OtpErlangRef senderRef = null;
+		try {
+			OtpErlangTuple call = (OtpErlangTuple) msg.getMsg();
+			OtpErlangTuple from = (OtpErlangTuple) call.elementAt(1);
+			request = (OtpErlangTuple) call.elementAt(2);
+			senderPid = (OtpErlangPid) from.elementAt(0);
+			senderRef = (OtpErlangRef) from.elementAt(1);
+			String module = ((OtpErlangAtom) request.elementAt(1)).atomValue();
+			String function = ((OtpErlangAtom) request.elementAt(2))
+					.atomValue();
+			OtpErlangObject args = request.elementAt(3);
+			OtpErlangList argsList = null;
+			// normalize input
+			if (args instanceof OtpErlangList) {
+				argsList = (OtpErlangList) args;
+			} else {
+				argsList = new OtpErlangList(args);
+			}
+			if (!nodeElement.getBinding().getModule().equals(module)) {
+				// module not found
+				// TODO: externalize message?
+				OtpErlangObject errorMsg = MessageHelper.functionUndefMessage(
+						module, function, argsList,
+						"Module not found in SCA component.");
+				sendMessage(connection, senderPid, senderRef,
+						MessageHelper.ATOM_BADRPC, errorMsg);
+			} else {
+				// module found, looking for operation
+				RuntimeComponentService service = nodeElement.getService();
+				ErlangBinding binding = nodeElement.getBinding();
+				List<Operation> operations = service.getInterfaceContract()
+						.getInterface().getOperations();
+				Operation operation = null;
+				for (Operation o : operations) {
+					if (o.getName().equals(function)) {
+						operation = o;
+						break;
+					}
+				}
+				if (operation != null) {
+					// operation found
+					List<DataType> iTypes = operation.getInputType()
+							.getLogical();
+					Class<?>[] forClasses = new Class<?>[iTypes.size()];
+					for (int i = 0; i < iTypes.size(); i++) {
+						forClasses[i] = iTypes.get(i).getPhysical();
+					}
+					try {
+						// invoke operation
+						Method jmethod = ((JavaOperation) operation)
+								.getJavaMethod();
+						Object result = service.getRuntimeWire(binding,
+								service.getInterfaceContract()).invoke(
+								operation,
+								TypeHelpersProxy.toJavaFromList(argsList,
+										forClasses, jmethod
+												.getParameterAnnotations()));
+						OtpErlangObject response = null;
+
+						// send reply
+						if (operation.getOutputType() != null
+								&& operation.getOutputType().getPhysical()
+										.isArray()) {
+							// output type is array
+							response = TypeHelpersProxy.toErlangAsResultList(
+									result, jmethod.getAnnotations());
+						} else if (operation.getOutputType() == null) {
+							// output type is void, create empty reply
+							Object[] arrArg = new Object[] {};
+							response = TypeHelpersProxy.toErlang(arrArg,
+									new Annotation[0][0]);
+						} else {
+							// output type is not void and not array
+							response = TypeHelpersProxy.toErlang(result,
+									jmethod.getAnnotations());
+						}
+						sendMessage(connection, senderPid, senderRef, null,
+								response);
+					} catch (Exception e) {
+						if ((e.getClass().equals(
+								InvocationTargetException.class) && e
+								.getCause().getClass().equals(
+										IllegalArgumentException.class))
+								|| e.getClass().equals(
+										TypeMismatchException.class)) {
+							// wrong params
+							// TODO: externalize message?
+							OtpErlangObject errorMsg = MessageHelper
+									.functionUndefMessage(module, function,
+											argsList,
+											"Operation name found in SCA component, but parameters types didn't match.");
+							sendMessage(connection, senderPid, senderRef,
+									MessageHelper.ATOM_BADRPC, errorMsg);
+						} else {
+							// unexpected error
+							throw e;
+						}
+					}
+				} else {
+					// operation not found
+					// TODO: externalize message?
+					OtpErlangObject errorMsg = MessageHelper
+							.functionUndefMessage(module, function, argsList,
+									"Operation name not found in SCA component.");
+					sendMessage(connection, senderPid, senderRef,
+							MessageHelper.ATOM_BADRPC, errorMsg);
+				}
+			}
+		} catch (ClassCastException e) {
+			// invalid request
+			// TODO: externalize message?
+			try {
+				logger
+						.log(
+								Level.WARNING,
+								"On node '"
+										+ nodeElement.getBinding().getNode()
+										+ "' received RPC request which is invalid. Request content is: "
+										+ msg.getMsg());
+			} catch (OtpErlangDecodeException e1) {
+			}
+		} catch (Exception e) {
+			// unknown error
+			try {
+				sendMessage(connection, senderPid, senderRef,
+						MessageHelper.ATOM_ERROR, new OtpErlangString(
+								"Unhandled error while processing request: "
+										+ e.getClass().getCanonicalName()
+										+ ", message: " + e.getMessage()));
+			} catch (Exception e1) {
+				// error while sending error message. Can't do anything now
+				logger.log(Level.WARNING, "Error during sending error message",
+						e);
+			}
+		}
+	}
+
+	private void handleMsg(OtpMsg msg) {
+		Operation matchedOperation = null;
+		Object args[] = null;
+		OtpErlangPid senderPid = null;
+		OtpErlangObject msgNoSender = null;
+		List<Operation> operations = groupedOperations.get(msg
+				.getRecipientName());
+		try {
+			if (msg.getMsg().getClass().equals(OtpErlangTuple.class)
+					&& (((OtpErlangTuple) msg.getMsg()).elementAt(0))
+							.getClass().equals(OtpErlangPid.class)) {
+				// PID provided by client
+				senderPid = (OtpErlangPid) ((OtpErlangTuple) msg.getMsg())
+						.elementAt(0);
+				msgNoSender = ((OtpErlangTuple) msg.getMsg()).elementAt(1);
+			} else {
+				// PID obtained from jinterface
+				senderPid = msg.getSenderPid();
+				msgNoSender = msg.getMsg();
+			}
+		} catch (Exception e) {
+			logger.log(Level.WARNING, "Unexpected error", e);
+		}
+
+		if (operations == null) {
+			// operation name not found
+			// TODO: externalize message?
+			// NOTE: I assume in Erlang sender doesn't get confirmation so
+			// no message will be send
+			logger.log(Level.WARNING, "Node '" + name
+					+ "' received message addressed to non exising mbox: "
+					+ msg.getRecipientName());
+		} else {
+			// find proper operation for received parameters
+			for (Operation operation : operations) {
+				Method method = ((JavaOperation) operation).getJavaMethod();
+				List<DataType> iTypes = operation.getInputType().getLogical();
+				Class<?>[] forClasses = new Class<?>[iTypes.size()];
+				for (int i = 0; i < iTypes.size(); i++) {
+					forClasses[i] = iTypes.get(i).getPhysical();
+				}
+				try {
+					args = TypeHelpersProxy.toJavaAsArgs(msgNoSender,
+							forClasses, method.getParameterAnnotations());
+					matchedOperation = operation;
+					break;
+				} catch (Exception e) {
+					// this exception is expected while processing operation
+					// version with mismatched arguments
+				}
+			}
+			if (matchedOperation != null) {
+				// operation found, invoke it
+				try {
+					Method jmethod = ((JavaOperation) matchedOperation)
+							.getJavaMethod();
+					Object result = nodeElement.getService().getRuntimeWire(
+							nodeElement.getBinding()).invoke(matchedOperation,
+							args);
+					OtpErlangObject response = null;
+
+					// create and send send reply
+					if (matchedOperation.getOutputType() != null
+							&& matchedOperation.getOutputType().getPhysical()
+									.isArray()) {
+						// result type is array
+						response = TypeHelpersProxy.toErlangAsResultList(
+								result, jmethod.getAnnotations());
+					} else if (matchedOperation.getOutputType() != null) {
+						// result type is not array and not void
+						response = TypeHelpersProxy.toErlang(result, jmethod
+								.getAnnotations());
+					}
+					if (response != null && senderPid != null) {
+						connection.send(senderPid, response);
+					} else if (response != null && senderPid == null) {
+						// couldn't send reply - sender pid unavailable
+						// TODO: externalize message?
+						// TODO: do we need to send this reply?
+						logger
+								.log(
+										Level.WARNING,
+										"Cannot send reply - Erlang client didn't provide it's PID and couldn't obtain sender PID from jinterface");
+					}
+				} catch (InvocationTargetException e) {
+					if (e.getCause() != null
+							&& e.getCause().getClass().equals(
+									IllegalArgumentException.class)) {
+						// arguments number or type mismatch
+						try {
+							// TODO: externalize message?
+							connection
+									.send(
+											senderPid,
+											new OtpErlangString(
+													"Operation name found in SCA component, but parameters types didn't match."));
+						} catch (IOException e1) {
+							e1.printStackTrace();
+						}
+					} else {
+						logger.log(Level.WARNING, "Unexpected error", e);
+					}
+				} catch (Exception e) {
+					logger.log(Level.WARNING, "Unexpected error", e);
+				}
+			} else {
+				// 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. Recevied arguments: " + msgNoSender);
+			}
+		}
+	}
+
+	public void run() {
+		try {
+			// NOTE: there's also a timeout, like in reference bindings
+			OtpMsg msg = null;
+			if (nodeElement.getBinding().hasTimeout()) {
+				msg = connection.receiveMsg(nodeElement.getBinding()
+						.getTimeout());
+			} else {
+				msg = connection.receiveMsg();
+			}
+			// check if request is message or RPC
+			if (msg.getRecipientName().equals(MessageHelper.RPC_MBOX)
+					&& !nodeElement.getBinding().isMbox()) {
+				handleRpc(msg);
+			} else if (!msg.getRecipientName().equals(MessageHelper.RPC_MBOX)
+					&& nodeElement.getBinding().isMbox()) {
+				handleMsg(msg);
+			} else {
+				// received wrong message type
+			}
+		} catch (IOException e) {
+			// TODO: externalize message?
+			logger.log(Level.WARNING, "Problem while receiving message", e);
+		} catch (OtpErlangExit e) {
+			// TODO: linking?
+		} catch (OtpAuthException e) {
+			// TODO: cookies? does this exception occur sometime?
+		} catch (InterruptedException e) {
+			// NOTE: timeout will be logged
+			// TODO: externalize message?
+			logger.log(Level.WARNING, "Timeout while waiting for request", e);
+		} finally {
+			connection.close();
+		}
+	}
+}