TransportServices.java

Go to the documentation of this file.

/*
 *                         Vortex OpenSplice
 *
 *   This software and documentation are Copyright 2006 to 2021 ADLINK
 *   Technology Limited, its affiliated companies and licensors. All rights
 *   reserved.
 *
 *   Licensed 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.vortex.FACE;

import java.io.File;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Level;

import org.omg.dds.core.AlreadyClosedException;
import org.omg.dds.core.IllegalOperationException;
import org.omg.dds.core.NotEnabledException;
import org.omg.dds.core.PreconditionNotMetException;
import org.omg.dds.core.ServiceEnvironment;

import FACE.CONNECTION_DIRECTION_TYPEHolder;
import FACE.MESSAGING_PATTERN_TYPE;
import FACE.RETURN_CODE_TYPE;
import FACE.RETURN_CODE_TYPEHolder;
import FACE.TRANSPORT_CONNECTION_STATUS_TYPEHolder;

public class TransportServices {
    private ServiceEnvironment env;
    private Map<String, ConnectionDescription> connectionDescriptions;
    private Map<Long, Connection<?>> connections;

    TransportServices() {
    }

    public void Initialize(String configuration,
            RETURN_CODE_TYPEHolder return_code) {
        System.setProperty(
                ServiceEnvironment.IMPLEMENTATION_CLASS_NAME_PROPERTY,
                "org.opensplice.dds.core.OsplServiceEnvironment");
        this.env = ServiceEnvironment.createInstance(TransportServices.class.getClassLoader());
        this.connectionDescriptions = Collections
                .synchronizedMap(new HashMap<String, ConnectionDescription>());
        this.connections = Collections
                .synchronizedMap(new HashMap<Long, Connection<?>>());
        if (return_code == null) {
            Logger.getInstance().log("Invalid argument return_code", Level.SEVERE);
            return;
        }
        if (configuration == null) {
            return_code.value = FACE.RETURN_CODE_TYPE.INVALID_PARAM;
            Logger.getInstance().log("Invalid argument configuration", Level.SEVERE);
            return;
        }
        File f = new File(configuration);
        if(f.exists() && !f.isDirectory()) {
            try {
                ConnectionDescription cd = new ConnectionDescription(env,configuration);
                for (ConnectionDescription connection : cd.getConfigurations()) {
                    connectionDescriptions.put(connection.getName(), connection);
                }
                return_code.value = FACE.RETURN_CODE_TYPE.NO_ERROR;
            } catch (Exception e) {
                Logger.getInstance().log(e.getMessage(), Level.SEVERE);
                Logger.getInstance().log(e.toString(), Level.FINEST);
                return_code.value = FACE.RETURN_CODE_TYPE.INVALID_CONFIG;
            }
        } else {
            return_code.value = FACE.RETURN_CODE_TYPE.INVALID_PARAM;
            Logger.getInstance().log("Invalid argument configuration, configuration file (" + f.getAbsolutePath() + ") does not exists", Level.SEVERE);
            return;
        }
    }


    public void Create_Connection(
            /* in */String connection_name,
            /* in */MESSAGING_PATTERN_TYPE pattern,
            /* out */us.opengroup.FACE.LongHolder connection_id,
            /* out */CONNECTION_DIRECTION_TYPEHolder connection_direction,
            /* out */us.opengroup.FACE.IntHolder max_message_size,
            /* in */long timeout,
            /* out */RETURN_CODE_TYPEHolder return_code) {
        if (return_code == null) {
            Logger.getInstance().log("Invalid argument return_code", Level.SEVERE);
            return;
        }
        if(connection_name == null ) {
            Logger.getInstance().log("Invalid argument connection_name", Level.SEVERE);
            return_code.value = FACE.RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        if(connection_id == null ) {
            Logger.getInstance().log("Invalid argument connection_id", Level.SEVERE);
            return_code.value = FACE.RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        if (connection_direction == null) {
            Logger.getInstance().log("Invalid argument connection_direction", Level.SEVERE);
            return_code.value = FACE.RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        if (pattern != MESSAGING_PATTERN_TYPE.PUB_SUB) {
            Logger.getInstance().log("Only PUB_SUB message type supported ", Level.SEVERE);
            return_code.value = FACE.RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        synchronized (this.connections) {
            ConnectionDescription description = connectionDescriptions.get(connection_name.toLowerCase());
            if (description == null) {
                Logger.getInstance().log("No connection found with name " + connection_name, Level.SEVERE);
                return_code.value = FACE.RETURN_CODE_TYPE.INVALID_CONFIG;
                return;
            }
            Connection<?> connection = this.connections.get(description.getId());
            if (connection != null) {
                return_code.value = FACE.RETURN_CODE_TYPE.NO_ACTION;
                connection_id.value = description.getId();
                connection_direction.value = description.getDirection();
            } else {
                try {
                    connection = Connection.getConnection(description,
                            Class.forName(description.getTypeName()));

                    connection_id.value = description.getId();
                    connection_direction.value = description.getDirection();

                    this.connections.put(connection_id.value, connection);

                    return_code.value = FACE.RETURN_CODE_TYPE.NO_ERROR;
                } catch (ClassNotFoundException e) {
                    Logger.getInstance()
                    .log("Class '"
                            + description.getTypeName()
                            + "'  for configured type_name could not be found. Please use fully qualified class name",
                            Level.SEVERE);
                    return_code.value = FACE.RETURN_CODE_TYPE.INVALID_CONFIG;
                } catch (Exception e) {
                    Logger.getInstance().log(e.getMessage(), Level.SEVERE);
                    Logger.getInstance().log(e.toString(), Level.FINEST);
                    return_code.value = this.getFACEReturnCodeFromException(e);
                }
            }
        }
        return;
    }

    public void Destroy_Connection(long connection_id,
            /* out */RETURN_CODE_TYPEHolder return_code) {
        if (return_code == null) {
            Logger.getInstance().log("Invalid argument return_code", Level.SEVERE);
            return;
        }
        synchronized (this.connections) {
            Connection<?> connection = this.connections.remove(connection_id);
            if (connection == null) {
                Logger.getInstance().log("Invalid argument connection_id could not find a connection for connection_id:" + connection_id, Level.SEVERE);
                return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
                return;
            }
            try {
                connection.close();
                return_code.value = RETURN_CODE_TYPE.NO_ERROR;
            } catch (Exception e) {
                Logger.getInstance().log(e.getMessage(), Level.SEVERE);
                return_code.value = RETURN_CODE_TYPE.INVALID_MODE;
            }
        }
        return;
    }

    public <TYPE> void Receive_Message(long connection_id, long timeout,
            us.opengroup.FACE.LongHolder transaction_id, Holder<TYPE> message,
            int message_size, RETURN_CODE_TYPEHolder return_code) {
        if (return_code == null) {
            Logger.getInstance().log("Invalid argument return_code", Level.SEVERE);
            return;
        }
        if (transaction_id == null) {
            Logger.getInstance().log("Invalid argument transaction_id", Level.SEVERE);
            return;
        }
        if (message == null) {
            Logger.getInstance().log("Invalid argument message", Level.SEVERE);
            return;
        }
        synchronized (this.connections) {
            Connection<?> connection = connections.get(connection_id);
            if (connection == null) {
                Logger.getInstance().log("Invalid argument connection_id could not find a connection for connection_id:" + connection_id, Level.SEVERE);
                return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
                return;
            }

            if (connection.getDirection() == FACE.CONNECTION_DIRECTION_TYPE.SOURCE) {
                return_code.value = RETURN_CODE_TYPE.INVALID_MODE;
                Logger.getInstance().log("Invalid connection direction for Receive_Message function", Level.SEVERE);
                return;
            }

            Connection<TYPE> typedConnection = connection.cast();

            if (typedConnection == null) {
                return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
                return;
            }
            DestinationConnection<TYPE> destConnection = typedConnection.asDestination();

            if (destConnection == null) {
                return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
                return;
            }
            destConnection.receiveMessage(timeout, transaction_id, message,
                    message_size, return_code);
        }
    }

    public <TYPE> void Send_Message(long connection_id, long timeout,
            us.opengroup.FACE.LongHolder transaction_id, Holder<TYPE> message, us.opengroup.FACE.IntHolder message_size,
            RETURN_CODE_TYPEHolder return_code) {
        if (return_code == null) {
            Logger.getInstance().log("Invalid argument return_code", Level.SEVERE);
            return;
        }
        if (transaction_id == null) {
            Logger.getInstance().log("Invalid argument transaction_id", Level.SEVERE);
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        if (message == null) {
            Logger.getInstance().log("Invalid argument message", Level.SEVERE);
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        synchronized (this.connections) {
            Connection<?> connection = connections.get(connection_id);

            if (connection == null) {
                Logger.getInstance().log("Invalid argument connection_id could not find a connection for connection_id:" + connection_id, Level.SEVERE);
                return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
                return;
            }

            if (connection.getDirection() == FACE.CONNECTION_DIRECTION_TYPE.DESTINATION) {
                return_code.value = RETURN_CODE_TYPE.INVALID_MODE;
                Logger.getInstance().log("Invalid connection direction for Send_Message function", Level.SEVERE);
                return;
            }

            Connection<TYPE> typedConnection = connection.cast();

            if (typedConnection == null) {
                return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
                return;
            }
            SourceConnection<TYPE> srcConnection = typedConnection.asSource();

            if (srcConnection == null) {
                return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
                return;
            }
            return_code.value = srcConnection.sendMessage(message.value, timeout,transaction_id);
        }

    }

    public <TYPE> void Register_Callback(
            /* in */long connection_id,
            /* in */boolean[] waitset,
            /* inout */Read_CallbackHolder<TYPE> data_callback,
            /* in */int max_message_size,
            /* out */RETURN_CODE_TYPEHolder return_code) {
        if (return_code == null) {
            Logger.getInstance().log("Invalid argument return_code", Level.SEVERE);
            return;
        }
        if (data_callback == null) {
            Logger.getInstance().log("Invalid argument data_callback", Level.SEVERE);
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        Connection<?> connection = connections.get(connection_id);

        if (connection == null) {
            Logger.getInstance().log("Invalid argument connection_id could not find a connection for connection_id:" + connection_id, Level.SEVERE);
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        if (connection.getDirection() == FACE.CONNECTION_DIRECTION_TYPE.SOURCE) {
            return_code.value = RETURN_CODE_TYPE.INVALID_MODE;
            Logger.getInstance().log("Invalid connection direction for Register_Callback function", Level.SEVERE);
            return;
        }

        Connection<TYPE> typedConnection = connection.cast();

        if (typedConnection == null) {
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        DestinationConnection<TYPE> destConnection = typedConnection
                .asDestination();

        if (destConnection == null) {
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        if (destConnection.getCallback() != null) {
            return_code.value = RETURN_CODE_TYPE.NO_ACTION;
            return;
        }
        return_code.value = destConnection.registerCallback(data_callback,
                max_message_size);
        return;
    }

    public <TYPE> void Unregister_Callback(long connection_id,
            RETURN_CODE_TYPEHolder return_code) {
        if (return_code == null) {
            Logger.getInstance().log("Invalid argument return_code", Level.SEVERE);
            return;
        }
        Connection<?> connection = connections.get(connection_id);

        if (connection == null) {
            Logger.getInstance().log("Invalid argument connection_id could not find a connection for connection_id:" + connection_id, Level.SEVERE);
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }

        if (connection.getDirection() == FACE.CONNECTION_DIRECTION_TYPE.SOURCE) {
            return_code.value = RETURN_CODE_TYPE.INVALID_MODE;
            Logger.getInstance().log("Invalid connection direction for Unregister_Callback function", Level.SEVERE);
            return;
        }

        Connection<TYPE> typedConnection = connection.cast();

        if (typedConnection == null) {
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        DestinationConnection<TYPE> destConnection = typedConnection
                .asDestination();

        if (destConnection == null) {
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        return_code.value = destConnection.unregisterCallback();

        return;

    }

    public void Get_Connection_Parameters(
            /* inout */us.opengroup.FACE.StringHolder connection_name,
            /* inout */us.opengroup.FACE.LongHolder connection_id,
            /* out */TRANSPORT_CONNECTION_STATUS_TYPEHolder connection_status,
            /* out */RETURN_CODE_TYPEHolder return_code) {
        if (return_code == null) {
            Logger.getInstance().log("Invalid argument return_code", Level.SEVERE);
            return;
        }
        if (connection_name == null) {
            Logger.getInstance().log("Invalid argument connection_name", Level.SEVERE);
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        if (connection_id == null) {
            Logger.getInstance().log("Invalid argument connection_id", Level.SEVERE);
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        if (connection_status == null) {
            Logger.getInstance().log("Invalid argument connection_status", Level.SEVERE);
            return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
            return;
        }
        synchronized (this.connections) {
            Connection<?> connection = connections.get(connection_id.value);

            if (connection == null) {
                Logger.getInstance().log("Invalid argument connection_id could not find a connection for connection_id:" + connection_id.value, Level.SEVERE);
                return_code.value = RETURN_CODE_TYPE.INVALID_PARAM;
                return;
            }
            connection_name.value = connection.getDescription().getName();
            connection_status.value = connection.getStatus();
            return_code.value = RETURN_CODE_TYPE.NO_ERROR;
        }
        return;
    }

    private RETURN_CODE_TYPE getFACEReturnCodeFromException(Exception e) {
        if (e instanceof AlreadyClosedException
                || e instanceof java.util.concurrent.TimeoutException) {
            return RETURN_CODE_TYPE.NO_ACTION;
        } else if (e instanceof UnsupportedOperationException) {
            return RETURN_CODE_TYPE.NOT_AVAILABLE;
        } else if (e instanceof IllegalArgumentException) {
            return RETURN_CODE_TYPE.INVALID_PARAM;
        } else if (e instanceof NotEnabledException
                || e instanceof PreconditionNotMetException
                || e instanceof IllegalOperationException) {
            return RETURN_CODE_TYPE.INVALID_MODE;
        }
        return RETURN_CODE_TYPE.INVALID_PARAM;

        /*
         * case DDS.RETCODE_ALREADY_DELETED.value: case
         * DDS.RETCODE_NO_DATA.value: return RETURN_CODE_TYPE.NO_ACTION; case
         * DDS.RETCODE_OK.value: return RETURN_CODE_TYPE.NO_ERROR; case
         * DDS.RETCODE_UNSUPPORTED.value: return RETURN_CODE_TYPE.NOT_AVAILABLE;
         * case DDS.RETCODE_BAD_PARAMETER.value: return
         * RETURN_CODE_TYPE.INVALID_PARAM; case DDS.RETCODE_TIMEOUT.value:
         * return RETURN_CODE_TYPE.TIMED_OUT; case
         * DDS.RETCODE_NOT_ENABLED.value: case
         * DDS.RETCODE_PRECONDITION_NOT_MET.value: case
         * DDS.RETCODE_ILLEGAL_OPERATION.value: return
         * RETURN_CODE_TYPE.INVALID_MODE; case DDS.RETCODE_ERROR.value: case
         * DDS.RETCODE_OUT_OF_RESOURCES.value: case
         * DDS.RETCODE_IMMUTABLE_POLICY.value: case
         * DDS.RETCODE_INCONSISTENT_POLICY.value: default: return
         * RETURN_CODE_TYPE.INVALID_CONFIG; }
         */
    }
}