Solace integration plugin

dnf install solace-plugin-*
sudo /etc/sysconfig/deephaven/plugins/solace-plugin/bin/activate.sh --classpath

rm -rf /etc/sysconfig/deephaven/plugins/solace-plugin
mkdir -p /etc/sysconfig/deephaven/plugins/solace-plugin
tar -xvf /tmp/solace-plugin/Solace-plugin-1.20200331.150_1.0-Manual.tgz -C /etc/sysconfig/deephaven/plugins/solace-plugin
/etc/sysconfig/deephaven/plugins/solace-plugin/bin/reinstall.sh

sudo mkdir -p /db/dataImportServers/SolaceImport1
sudo chown dbquery.dbmergegrp /db/dataImportServers/SolaceImport1

builder = new JSONToTableWriterAdapter.Builder()
    .allowUnmapped("Status")
    .allowUnmapped("Event")
    .addColumnToFieldMapping("OriginalTimestamp", "Timestamp")
    .messageIdColumnName("MessageId")
    .timestampColumnName("IngestTimestamp")
    .solaceReceiveTimestampColumnName("SolaceReceiveTimestamp")
    .solaceSendTimestampColumnName("SolaceSendTimestamp")
    .allowNullValues(true)

builder = new SimpleStringToTableWriterAdapter.Builder()
    .allowUnmapped("Status")
    .allowUnmapped("Event")
    .setValueColumnName("RawText")
    .messageIdColumnName("MessageId")
    .timestampColumnName("IngestTimestamp")
    .solaceReceiveTimestampColumnName("SolaceReceiveTimestamp")
    .solaceSendTimestampColumnName("SolaceSendTimestamp")
    .allowNullValues(true)

def notionalFunction = { JsonRecord record ->
    int size = JsonRecordUtil.getInt(record, "Size")
    double price = JsonRecordUtil.getDouble(record, "Price")
    return (size * price)
}

def conditionSplit = { JsonRecord record ->
    String condition = JsonRecordUtil.getString(record, "SaleCondition")
    if (condition == null) {
        return null;
    }
    return condition.split(",")
}

builder = new JSONToTableWriterAdapter.Builder()
    .allowUnmapped("Status")
    .allowUnmapped("Event")
    .addColumnToFieldMapping("OriginalTimestamp", "Timestamp")
    .messageIdColumnName("MessageId")
    .timestampColumnName("IngestTimestamp")
    .solaceReceiveTimestampColumnName("SolaceReceiveTimestamp")
    .solaceSendTimestampColumnName("SolaceSendTimestamp")
    .addColumnToDoubleFunction("Notional", notionalFunction)
    .addColumnToFunction("Conditions", String[].class, conditionSplit)
    .allowNullValues(true)

// Deephaven imports

import com.illumon.iris.db.tables.dataimport.logtailer.DataImportServer;
import com.fishlib.configuration.Configuration;
import com.fishlib.util.process.ProcessEnvironment;
import com.illumon.iris.db.util.logging.IrisLogCreator;
import com.illumon.iris.db.v2.routing.DataRoutingServiceFactory;
import io.deephaven.solace.ingester.SolaceQueueIngester;
import io.deephaven.solace.ingester.JSONToTableWriterAdapter;
import com.solacesystems.jcsmp.Consumer;
import com.solacesystems.jcsmp.*;

// Copy the host, vpn, username and password from the Deephaven
// configuration.
// You may want to hardcode these values, or alternatively can set them in
// the Deephaven configuration or by using System properties from the
// 'Extra JVM arguments' panel.
final JCSMPProperties properties = new JCSMPProperties();
properties.setProperty(JCSMPProperties.HOST, Configuration.getInstance().getProperty("SolaceQueueIngester.host"));
properties.setProperty(JCSMPProperties.USERNAME, Configuration.getInstance().getProperty("SolaceQueueIngester.username"));
properties.setProperty(JCSMPProperties.VPN_NAME, Configuration.getInstance().getProperty("SolaceQueueIngester.vpn"));
properties.setProperty(JCSMPProperties.PASSWORD, Configuration.getInstance().getProperty("SolaceQueueIngester.password"));
properties.setProperty(JCSMPProperties.GENERATE_RCV_TIMESTAMPS, true);

final JCSMPSession session = JCSMPFactory.onlyInstance().createSession(properties);
session.connect();

final EndpointProperties endpointProps = new EndpointProperties();
endpointProps.setPermission(EndpointProperties.PERMISSION_CONSUME);
endpointProps.setAccessType(EndpointProperties.ACCESSTYPE_EXCLUSIVE);

// In this example we are using a “trades” queue that contains market trades
// in JSON format, with the 'SolaceQueueIngester.queueName' configuration variable set to the
// value "trades".
final com.solacesystems.jcsmp.Queue queue = JCSMPFactory.onlyInstance().createQueue(Configuration.getInstance().getProperty("SolaceQueueIngester.queueName"));
session.provision(queue, endpointProps, JCSMPSession.FLAG_IGNORE_ALREADY_EXISTS);
final ConsumerFlowProperties flow_prop = new ConsumerFlowProperties();
flow_prop.setEndpoint(queue);
flow_prop.setAckMode(JCSMPProperties.SUPPORTED_MESSAGE_ACK_CLIENT);
EndpointProperties endpoint_props = new EndpointProperties();
endpoint_props.setAccessType(EndpointProperties.ACCESSTYPE_EXCLUSIVE);


// create the DataImportServer and its prerequisites
routingService = DataRoutingServiceFactory.getDefault();

// if the dis configuration specifies "private" storage, then you must provide a storage location, otherwise set this to null
storage = "/db/dataImportServers/SolaceImport1"
// the SolaceImport1 string is the name given to this in-worker DataImportServer configuration
disName = "SolaceImport1"

dis = DataImportServer.getDataImportServer(null, disName, Configuration.getInstance(), db.getSchemaService(), routingService, storage)
dis.startInWorker();

// By default the JSONToTableWriterAdapter maps columns to JSON fields of
// the same name.  Additionally, we set the values of “Status” and “Event” to
// null, because they do not exist in our JSON.  We use the “Timestamp” field
// within the JSON records to populate the “OriginalTimestamp” column.  The
// MessageID column is populated from the Solace message properties [this // can be useful for debugging to trace the provenance of a row of data], and
// the IngestTimestamp is the wall-clock time of the ingestion.  We allow the
// JSON records to contain null values.
builder = new JSONToTableWriterAdapter.Builder()
    .allowUnmapped("Status")
    .allowUnmapped("Event")
    .addColumnToFieldMapping("OriginalTimestamp", "Timestamp")
    .messageIdColumnName("MessageId")
       .timestampColumnName("IngestTimestamp")
// note that if the Solace library does not populate this field. NULL is written to Deephaven
       .solaceReceiveTimestampColumnName("SolaceReceiveTimestamp")
       .solaceSendTimestampColumnName("SolaceSendTimestamp")
       .allowNullValues(true)

// Create a SolaceQueueIngester with an ingestername of “TradeImport”,
// with a target of the Solace.TradeNbboStock table.  Each Solace message
// must have the property “MsgId” which is a unique increasing String value
// that is used to reliably acknowledge messages once they have been
// durably committed to Deephaven.  The final `builder.buildFactory()`
// argument is the definition above of how to process the JSON messages.
{{ ... }}
si = new SolaceQueueIngester(log, "TradeImport", "Solace", "TradeNbboStock", currentDateNy(), "MsgId", session, queue, flow_prop, endpoint_props, dis, builder.buildFactory(log))
si.start()

// The table is now available from this and other workers
jsonTable=db.i("Solace", "TradeNbboStock").where("Date=currentDateNy()")
{{ ... }}

builder = new PojoToTableWriterAdapter.Builder()
    .valueClass(SerializedTestClass.class)
    .caseInsensitiveSearch(true)
    .allowUnmapped("Status")
    .allowUnmapped("Event")
    .addColumnToValueField("OriginalTimestamp", "Timestamp")
    .addColumnToValueMethod("OrderNo", "getOrderNumber")
    .messageIdColumnName("MessageId")
    .timestampColumnName("IngestTimestamp")
    .solaceReceiveTimestampColumnName("SolaceReceiveTimestamp")
    .solaceSendTimestampColumnName("SolaceSendTimestamp")

.addColumnToSetter("OrderDate","com.illumon.iris.db.tables.utils.DBTimeUtils.millisToTime(deephavenSolaceIngesterObjectVariable.getOrderMillis())")

public void doMapping(
    @NotNull final Map<String, String> columnToSetter,
    @NotNull final Class valueClass,
    @NotNull final Map<String, Class> columnNameToType,
    @NotNull final String valueObjectName,
    @NotNull final List<String> columnValueExpressions
);

import io.deephaven.solace.ingester.PojoAdapterColumnMapper;
import java.lang.reflect.Method;

private static class CustomArrayMapper implements PojoAdapterColumnMapper {
    private boolean methodExists(@NotNull final Class<?> inputClass,
            @NotNull final String methodName,
            final Class<?> parameterClass) {
        try {
            if (parameterClass != null) {
                final Method method = inputClass.getMethod(methodName, parameterClass);
            } else {
                final Method method = inputClass.getMethod(methodName);
            }
        } catch (NoSuchMethodException e) {
            return false;
        }
        return true;
    }

    private String capitalizeFirstLetter(@NotNull final String input) {
        if (input.length() < 2) {
            return input.toUpperCase();
        }
        return input.substring(0,1).toUpperCase() + input.substring(1);
    }

    public void doMapping(
            @NotNull final Map<String, String> columnToSetter,
            @NotNull final Class valueClass,
            @NotNull final Map<String, Class> columnNameToType,
            @NotNull final String valueObjectName,
            @NotNull final List<String> columnValueExpressions) {
        for (final String col : columnNameToType.keySet()) {
            Class<?> colType = columnNameToType.get(col);
            if (colType.isArray()) {
                final String lengthMethod = "get" + capitalizeFirstLetter(col) + "Length";
                final String elementMethod = "get" + capitalizeFirstLetter(col) + "At";

                if (methodExists(valueClass, lengthMethod, null)
                        && methodExists(valueClass, elementMethod, int.class)) {
                    final String length = valueObjectName + "." + lengthMethod + "()";
                    final String element = valueObjectName + "." + elementMethod + "(";
                    final String retval = "deephavenSolaceRetval" + col;
                    final String valueExpression = "final " + colType.getSimpleName() +
                            " " + retval + " = new " + colType.getComponentType().getSimpleName() + "[" + length +
                            "];\n for (int i=0; i<" + length + "; i++) {\n " + retval + "[i]=" + element +
                            "i);\n }\n";
                    columnValueExpressions.add(valueExpression);
                    columnToSetter.put(col, retval);
                }
            }
        }

}
}

// Deephaven imports

import com.illumon.iris.db.tables.dataimport.logtailer.DataImportServer;
import com.fishlib.configuration.Configuration;
import com.fishlib.util.process.ProcessEnvironment;
import com.illumon.iris.db.v2.configuration.DataRoutingServiceFactory;
import io.deephaven.solace.ingester.SolaceQueueIngester;
import io.deephaven.solace.ingester.PojoToTableWriterAdapter;
import com.solacesystems.jcsmp.*;

// Copy the host, vpn, username and password from the Deephaven
// configuration.
// You may want to hardcode these values, or alternatively can set them in
// the Deephaven configuration or by using System properties from the
// 'Extra JVM arguments' panel.
final JCSMPProperties properties = new JCSMPProperties();
properties.setProperty(JCSMPProperties.HOST, Configuration.getInstance().getProperty("SolaceQueueIngester.host"));
properties.setProperty(JCSMPProperties.USERNAME, Configuration.getInstance().getProperty("SolaceQueueIngester.username"));
properties.setProperty(JCSMPProperties.VPN_NAME, Configuration.getInstance().getProperty("SolaceQueueIngester.vpn"));
properties.setProperty(JCSMPProperties.PASSWORD, Configuration.getInstance().getProperty("SolaceQueueIngester.password"));
properties.setProperty(JCSMPProperties.GENERATE_RCV_TIMESTAMPS, true);

final JCSMPSession session = JCSMPFactory.onlyInstance().createSession(properties);
session.connect();

final EndpointProperties endpointProps = new EndpointProperties();
endpointProps.setPermission(EndpointProperties.PERMISSION_CONSUME);
endpointProps.setAccessType(EndpointProperties.ACCESSTYPE_EXCLUSIVE);

// In this example we are using an “orders” queue that contains market orders
// in proprietary format that will be interpreted by the io.deephaven.solace.OrderEvent
// class.
final com.solacesystems.jcsmp.Queue queue = JCSMPFactory.onlyInstance().createQueue(Configuration.getInstance().getProperty("SolaceQueueIngester.queueName"));
session.provision(queue, endpointProps, JCSMPSession.FLAG_IGNORE_ALREADY_EXISTS);
final ConsumerFlowProperties flow_prop = new ConsumerFlowProperties();
flow_prop.setEndpoint(queue);
flow_prop.setAckMode(JCSMPProperties.SUPPORTED_MESSAGE_ACK_CLIENT);
EndpointProperties endpoint_props = new EndpointProperties();
endpoint_props.setAccessType(EndpointProperties.ACCESSTYPE_EXCLUSIVE);

// create the DataImportServer and its prerequisites
routingService = DataRoutingServiceFactory.getDefault();

// the SolaceImport1 string is the name given to this in-worker
// DataImportServer within the routing YAML configuration file
disConfig = routingService.getDataImportServiceConfig("SolaceImport1");
if (disConfig == null) {
throw new IllegalArgumentException("Could not find disConfig!");
}
dis = new DataImportServer(ProcessEnvironment.getDefaultLog(), disConfig, Configuration.getInstance());
dis.startInWorker();

// By default the PojoToTableWriterAdapter maps columns to object methods or fields
// with the same name. Additionally, we set the value of “UnsuppliedColumnName” to
// null, because it does not exist in our class. The MessageID column is populated from
// the Solace message properties [this can be useful for debugging to trace the provenance
// of a row of data], and the IngestTimestamp is the wall-clock time of the ingestion.
orderAdapterFactory = new PojoToTableWriterAdapter.Builder()
    .allowUnmapped("UnsuppliedColumnName")
    .valueClass(OrderEvent.class)
    .valueObjectCreator("new OrderEventClass(metadata.getBytes());")
    .imports("io.deephaven.solace.OrderEvent.class")
    .caseInsensitiveSearch(true)
    .messageIdColumnName("MessageId")
    .timestampColumnName("IngestTimestamp")
    .buildFactory(log);

// Uncomment the next line if (extremely) verbose logging is needed
//log.setLevel(com.fishlib.io.log.LogLevel.TRACE)

si = new SolaceQueueIngester(log, "OrderEventsImport", "TestNamespace", "OrderEventsTable", currentDateNy(), "MsgId", true, session, queue, flow_prop, endpoint_props, dis, orderAdapterFactory);

si.start()

// The table is now available from this and other workers
pojoTable=db.i("TestNamespace", "OrderEventsTable").where("Date=currentDateNy()")

// A sample Persistent Query for messages being written to one table
// Deephaven imports

import com.illumon.iris.db.tables.dataimport.logtailer.DataImportServer;
import com.fishlib.configuration.Configuration;
import com.fishlib.util.process.ProcessEnvironment;
import com.illumon.iris.db.v2.configuration.DataRoutingServiceFactory;
import io.deephaven.solace.ingester.SolaceQueueIngester;
import io.deephaven.solace.ingester.PojoToTableWriterAdapter;
import com.solacesystems.jcsmp.*;
import io.deephaven.solace.ingester.BytesXMLMessageToTableWriterAdapter;
import io.deephaven.solace.ingester.EmbeddedObjectAdapter;

// Copy the host, vpn, username and password from the Deephaven
// configuration.
// You may want to hardcode these values, or alternatively can set them in
// the Deephaven configuration or by using System properties from the
// 'Extra JVM arguments' panel.
final JCSMPProperties properties = new JCSMPProperties();
properties.setProperty(JCSMPProperties.HOST, Configuration.getInstance().getProperty("SolaceQueueIngester.host"));
properties.setProperty(JCSMPProperties.USERNAME, Configuration.getInstance().getProperty("SolaceQueueIngester.username"));
properties.setProperty(JCSMPProperties.VPN_NAME, Configuration.getInstance().getProperty("SolaceQueueIngester.vpn"));
properties.setProperty(JCSMPProperties.PASSWORD, Configuration.getInstance().getProperty("SolaceQueueIngester.password"));
properties.setProperty(JCSMPProperties.GENERATE_RCV_TIMESTAMPS, true);

final JCSMPSession session = JCSMPFactory.onlyInstance().createSession(properties);
session.connect();

final EndpointProperties endpointProps = new EndpointProperties();
endpointProps.setPermission(EndpointProperties.PERMISSION_CONSUME);
endpointProps.setAccessType(EndpointProperties.ACCESSTYPE_EXCLUSIVE);

// In this example we are using an “orders” queue that contains market orders
// in proprietary format that will be interpreted by the io.deephaven.solace.OrderEvent
// class.
final com.solacesystems.jcsmp.Queue queue = JCSMPFactory.onlyInstance().createQueue(Configuration.getInstance().getProperty("SolaceQueueIngester.queueName"));
session.provision(queue, endpointProps, JCSMPSession.FLAG_IGNORE_ALREADY_EXISTS);
final ConsumerFlowProperties flow_prop = new ConsumerFlowProperties();
flow_prop.setEndpoint(queue);
flow_prop.setAckMode(JCSMPProperties.SUPPORTED_MESSAGE_ACK_CLIENT);
EndpointProperties endpoint_props = new EndpointProperties();
endpoint_props.setAccessType(EndpointProperties.ACCESSTYPE_EXCLUSIVE);

// create the DataImportServer and its prerequisites
routingService = DataRoutingServiceFactory.getDefault();

// the SolaceImport1 string is the name given to this in-worker
// DataImportServer within the routing YAML configuration file
disConfig = routingService.getDataImportServiceConfig("SolaceImport1");
if (disConfig == null) {
throw new IllegalArgumentException("Could not find disConfig!");
}
dis = new DataImportServer(ProcessEnvironment.getDefaultLog(), disConfig, Configuration.getInstance());
dis.startInWorker();

// Adapters that will write embedded objects to other tables must be created
// before the adapter for the top-level messages, because these sub-adapters
// have to be available when building the main adapter.
orderDetailAdapterFactory = new PojoToTableWriterAdapter.Builder()
    .imports("io.deephaven.solace.OrderDetail")
    .valueClass(OrderDetail.class)
    // The parent adapter retrieves a generic object from the parent adapter
    // and it must be cast into its valueClass in the valueObjectCreator of
    // this adapter.
    .valueObjectCreator("(OrderDetail)(metadata.getValueObject());")
    .addColumnToSetter("orderEventMessageId","metadata.getParentIdValue()")
    .caseInsensitiveSearch(true)
    .buildFactory(log);

// The sub-adapter itself is created using new EmbeddedObjectAdapter; for the
// top-level adapter, the adapter is created internally when creating the ingester,
// but, for sub-adapters, an explicit call is required to set up the data processor
// that will manage the table's files.
final BytesXMLMessageToTableWriterAdapter orderDetailAdapter = new EmbeddedObjectAdapter(
    "TestNamespace",
    "OrderDetailsTable",
    "NestedOrderDetailIngester",
    DBTimeUtils.currentDateNy(),
    orderDetailAdapterFactory,
    dis,
    log).getEmbeddedObjectAdapter();

// By default the PojoToTableWriterAdapter maps columns to object methods or fields
// with the same name. Additionally, we set the value of “UnsuppliedColumnName” to
// null, because it does not exist in our class. The MessageID column is populated from
// the Solace message properties [this can be useful for debugging to trace the provenance
// of a row of data], and the IngestTimestamp is the wall-clock time of the ingestion.
orderAdapterFactory = new PojoToTableWriterAdapter.Builder()
    .allowUnmapped("UnsuppliedColumnName")
    .valueClass(OrderEvent.class)
    .valueObjectCreator("new OrderEventClass(metadata.getBytes());")
    .imports("io.deephaven.solace.OrderEvent.class")
    .caseInsensitiveSearch(true)
    .messageIdColumnName("MessageId")
    .timestampColumnName("IngestTimestamp")
    .addObjectToAdapter("getOrderDetail()",orderDetailAdapter,"MessageId")
    .buildFactory(log);

// Uncomment the next line if (extremely) verbose logging is needed
//log.setLevel(com.fishlib.io.log.LogLevel.TRACE)

si = new SolaceQueueIngester(log, "OrderEventsImport", "TestNamespace", "OrderEventsTable", currentDateNy(), "MsgId", true, session, queue, flow_prop, endpoint_props, dis, orderAdapterFactory);

// One additional step that is needed with hierarchical ingesters is to add sub-adapters
// to the ingester. This is so state of checkpointed data can be tracked across child tables.
si.addAdapter(orderDetailAdapter);

si.start()

// The tables are now available from this and other workers
orderEventTable=db.i("TestNamespace", "OrderEventsTable").where("Date=currentDateNy()")
orderDetailTable=db.i("TestNamespace", "OrderDetailsTable").where("Date=currentDateNy()")