summaryrefslogtreecommitdiffstats
path: root/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ServiceExecutor.java
diff options
context:
space:
mode:
Diffstat (limited to 'sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ServiceExecutor.java')
-rw-r--r--sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ServiceExecutor.java262
1 files changed, 262 insertions, 0 deletions
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
new file mode 100644
index 0000000000..63c58cb696
--- /dev/null
+++ b/sandbox/wjaniszewski/binding-erlang-runtime/src/main/java/org/apache/tuscany/sca/binding/erlang/impl/ServiceExecutor.java
@@ -0,0 +1,262 @@
+/*
+ * 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.reflect.InvocationTargetException;
+import java.util.List;
+import java.util.Map;
+
+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.runtime.RuntimeComponentService;
+
+import com.ericsson.otp.erlang.OtpConnection;
+import com.ericsson.otp.erlang.OtpErlangAtom;
+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.OtpMbox;
+import com.ericsson.otp.erlang.OtpMsg;
+import com.ericsson.otp.erlang.OtpNode;
+
+/**
+ * @version $Rev$ $Date$
+ */
+public class ServiceExecutor implements Runnable {
+
+ private static final long RECEIVE_TIMEOUT = 60000;
+
+ private Map<String, ErlangNodeElement> erlangModules;
+ private ErlangNodeElement erlangMbox;
+ private OtpConnection connection;
+ private Map<String, List<Operation>> groupedOperations;
+
+ public ServiceExecutor(OtpConnection connection,
+ Map<String, List<Operation>> groupedOperations,
+ Map<String, ErlangNodeElement> erlangModules,
+ ErlangNodeElement erlangMbox) {
+ this.erlangModules = erlangModules;
+ this.connection = connection;
+ this.groupedOperations = groupedOperations;
+ this.erlangMbox = erlangMbox;
+ }
+
+ private void sendMessage(OtpConnection connection, OtpErlangPid pid,
+ OtpErlangRef ref, OtpErlangAtom head, OtpErlangObject message)
+ throws IOException {
+ OtpErlangTuple tResult = new OtpErlangTuple(new OtpErlangObject[] {
+ head, 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 (!erlangModules.containsKey(module)) {
+ // 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 {
+ RuntimeComponentService service = erlangModules.get(module)
+ .getService();
+ ErlangBinding binding = erlangModules.get(module).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) {
+ 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 {
+ Object result = service.getRuntimeWire(binding).invoke(
+ operation,
+ TypeHelpersProxy.toJavaFromList(argsList,
+ forClasses));
+ OtpErlangObject response = null;
+ if (operation.getOutputType() != null
+ && operation.getOutputType().getPhysical()
+ .isArray()) {
+ response = TypeHelpersProxy.toErlangAsList(result);
+ } else if (operation.getOutputType() == null) {
+ Object[] arrArg = new Object[] {};
+ response = TypeHelpersProxy.toErlang(arrArg);
+ } else {
+ Object[] arrArg = new Object[] { result };
+ response = TypeHelpersProxy.toErlang(arrArg);
+ }
+ sendMessage(connection, senderPid, senderRef,
+ MessageHelper.ATOM_OK, response);
+ } catch (Exception e) {
+ if ((e.getClass().equals(
+ InvocationTargetException.class) && e
+ .getCause().getClass().equals(
+ IllegalArgumentException.class))
+ || e.getClass().equals(
+ TypeMismatchException.class)) {
+ // 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 {
+ throw e;
+ }
+ }
+ } else {
+ // 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 (Exception e) {
+ // TODO: distinguish and describe errors!
+ try {
+ e.printStackTrace();
+ sendMessage(connection, senderPid, senderRef,
+ MessageHelper.ATOM_ERROR, new OtpErlangString(
+ "Unhandled error while processing request: "
+ + e.getClass().getCanonicalName()
+ + ", message: " + e.getMessage()));
+ } catch (IOException e1) {
+ // error while sending error message. Can't do anything now
+ }
+ }
+ }
+
+ private void handleMsg(OtpMsg msg) {
+ Operation matchedOperation = null;
+ Object args[] = null;
+ List<Operation> operations = groupedOperations.get(msg
+ .getRecipientName());
+ if (operations == null) {
+ // TODO: no such mbox, send error message?
+ } else {
+ for (Operation operation : operations) {
+ 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(msg.getMsg(),
+ forClasses);
+ matchedOperation = operation;
+ break;
+ } catch (Exception e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+ if (matchedOperation != null) {
+ try {
+ Object result = erlangMbox.getService().getRuntimeWire(
+ erlangMbox.getBinding()).invoke(matchedOperation,
+ args);
+ OtpErlangObject response = null;
+ if (matchedOperation.getOutputType() != null
+ && matchedOperation.getOutputType().getPhysical()
+ .isArray()) {
+ response = TypeHelpersProxy.toErlangAsList(result);
+ } else if (matchedOperation.getOutputType() != null) {
+ Object[] arrArg = new Object[] { result };
+ response = TypeHelpersProxy.toErlang(arrArg);
+ }
+ if (response != null) {
+ OtpNode node = new OtpNode("_response_connector_to_"
+ + msg.getSenderPid());
+ OtpMbox mbox = node.createMbox();
+ mbox.send(msg.getSenderPid(), response);
+ }
+ } catch (InvocationTargetException e) {
+ // TODO send some error?
+ e.printStackTrace();
+ } catch (IOException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ } else {
+ // TODO: send some error - no mapping for such arguments
+ System.out
+ .println("TODO: send some error - no mapping for such arguments");
+ }
+ }
+ }
+
+ public void run() {
+ try {
+ OtpMsg msg = connection.receiveMsg(RECEIVE_TIMEOUT);
+ if (msg.getRecipientName().equals(MessageHelper.RPC_MBOX)) {
+ handleRpc(msg);
+ } else if (msg != null) {
+ handleMsg(msg);
+ } else {
+ // message receive timeout
+ }
+ } catch (Exception e) {
+ // TODO: log, send error?
+ e.printStackTrace();
+ } finally {
+ connection.close();
+ }
+ }
+}
generated by cgit v1.2.3 (git 2.25.1) at 2024-05-23 21:18:34 +0000