integration/activemq-vertx-integration/src/main/java/org/apache/activemq/artemis/integration/vertx/IncomingVertxEventHandler.java - activemq-artemis - Git at Google

 /*
  * 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.activemq.artemis.integration.vertx;

 import java.util.Map;

 import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
 import org.apache.activemq.artemis.api.core.SimpleString;
 import org.apache.activemq.artemis.core.persistence.StorageManager;
 import org.apache.activemq.artemis.core.postoffice.Binding;
 import org.apache.activemq.artemis.core.postoffice.PostOffice;
 import org.apache.activemq.artemis.core.server.ConnectorService;
 import org.apache.activemq.artemis.core.server.ServerMessage;
 import org.apache.activemq.artemis.core.server.impl.ServerMessageImpl;
 import org.apache.activemq.artemis.utils.ConfigurationHelper;
 import org.vertx.java.core.Handler;
 import org.vertx.java.core.buffer.Buffer;
 import org.vertx.java.core.eventbus.EventBus;
 import org.vertx.java.core.eventbus.Message;
 import org.vertx.java.core.eventbus.ReplyException;
 import org.vertx.java.core.eventbus.impl.PingMessage;
 import org.vertx.java.core.json.JsonArray;
 import org.vertx.java.core.json.JsonObject;
 import org.vertx.java.platform.PlatformLocator;
 import org.vertx.java.platform.PlatformManager;
 import org.vertx.java.spi.cluster.impl.hazelcast.HazelcastClusterManagerFactory;

 class IncomingVertxEventHandler implements ConnectorService {

    private final String connectorName;

    private final String queueName;

    private final int port;

    private final String host;

    private final int quorumSize;

    private final String haGroup;

    private final String vertxAddress;

    private EventBus eventBus;

    private PlatformManager platformManager;

    private EventHandler handler;

    private final StorageManager storageManager;

    private final PostOffice postOffice;

    private boolean isStarted = false;

    IncomingVertxEventHandler(String connectorName,
                              Map<String, Object> configuration,
                              StorageManager storageManager,
                              PostOffice postOffice) {
       this.connectorName = connectorName;
       this.queueName = ConfigurationHelper.getStringProperty(VertxConstants.QUEUE_NAME, null, configuration);

       this.port = ConfigurationHelper.getIntProperty(VertxConstants.PORT, 0, configuration);
       this.host = ConfigurationHelper.getStringProperty(VertxConstants.HOST, "localhost", configuration);
       this.quorumSize = ConfigurationHelper.getIntProperty(VertxConstants.VERTX_QUORUM_SIZE, -1, configuration);
       this.haGroup = ConfigurationHelper.getStringProperty(VertxConstants.VERTX_HA_GROUP, "activemq", configuration);
       this.vertxAddress = ConfigurationHelper.getStringProperty(VertxConstants.VERTX_ADDRESS, "org.apache.activemq", configuration);

       this.storageManager = storageManager;
       this.postOffice = postOffice;
    }

    @Override
    public void start() throws Exception {
       if (this.isStarted) {
          return;
       }
       System.setProperty("vertx.clusterManagerFactory", HazelcastClusterManagerFactory.class.getName());
       if (quorumSize != -1) {
          platformManager = PlatformLocator.factory.createPlatformManager(port, host, quorumSize, haGroup);
       }
       else {
          platformManager = PlatformLocator.factory.createPlatformManager(port, host);
       }

       eventBus = platformManager.vertx().eventBus();

       Binding b = postOffice.getBinding(new SimpleString(queueName));
       if (b == null) {
          throw new Exception(connectorName + ": queue " + queueName + " not found");
       }

       handler = new EventHandler();
       eventBus.registerHandler(vertxAddress, handler);

       isStarted = true;
       ActiveMQVertxLogger.LOGGER.debug(connectorName + ": started");
    }

    @Override
    public void stop() throws Exception {
       if (!isStarted) {
          return;
       }
       eventBus.unregisterHandler(vertxAddress, handler);
       platformManager.stop();
       System.clearProperty("vertx.clusterManagerFactory");
       isStarted = false;
       ActiveMQVertxLogger.LOGGER.debug(connectorName + ": stopped");
    }

    @Override
    public boolean isStarted() {
       return isStarted;
    }

    @Override
    public String getName() {
       return connectorName;
    }

    private class EventHandler implements Handler<Message<?>> {

       @Override
       public void handle(Message<?> message) {
          ServerMessage msg = new ServerMessageImpl(storageManager.generateID(), VertxConstants.INITIAL_MESSAGE_BUFFER_SIZE);
          msg.setAddress(new SimpleString(queueName));
          msg.setDurable(true);
          msg.encodeMessageIDToBuffer();

          String replyAddress = message.replyAddress();
          if (replyAddress != null) {
             msg.putStringProperty(VertxConstants.VERTX_MESSAGE_REPLYADDRESS, replyAddress);
          }

          // it'd be better that Message expose its type information
          int type = getMessageType(message);

          msg.putIntProperty(VertxConstants.VERTX_MESSAGE_TYPE, type);

          manualEncodeVertxMessageBody(msg.getBodyBuffer(), message.body(), type);

          try {
             postOffice.route(msg, null, false);
          }
          catch (Exception e) {
             ActiveMQVertxLogger.LOGGER.error("failed to route msg " + msg, e);
          }
       }

       private void manualEncodeVertxMessageBody(ActiveMQBuffer bodyBuffer, Object body, int type) {
          switch (type) {
             case VertxConstants.TYPE_BOOLEAN:
                bodyBuffer.writeBoolean(((Boolean) body));
                break;
             case VertxConstants.TYPE_BUFFER:
                Buffer buff = (Buffer) body;
                int len = buff.length();
                bodyBuffer.writeInt(len);
                bodyBuffer.writeBytes(((Buffer) body).getBytes());
                break;
             case VertxConstants.TYPE_BYTEARRAY:
                byte[] bytes = (byte[]) body;
                bodyBuffer.writeInt(bytes.length);
                bodyBuffer.writeBytes(bytes);
                break;
             case VertxConstants.TYPE_BYTE:
                bodyBuffer.writeByte((byte) body);
                break;
             case VertxConstants.TYPE_CHARACTER:
                bodyBuffer.writeChar((Character) body);
                break;
             case VertxConstants.TYPE_DOUBLE:
                bodyBuffer.writeDouble((double) body);
                break;
             case VertxConstants.TYPE_FLOAT:
                bodyBuffer.writeFloat((Float) body);
                break;
             case VertxConstants.TYPE_INT:
                bodyBuffer.writeInt((Integer) body);
                break;
             case VertxConstants.TYPE_LONG:
                bodyBuffer.writeLong((Long) body);
                break;
             case VertxConstants.TYPE_SHORT:
                bodyBuffer.writeShort((Short) body);
                break;
             case VertxConstants.TYPE_STRING:
             case VertxConstants.TYPE_PING:
                bodyBuffer.writeString((String) body);
                break;
             case VertxConstants.TYPE_JSON_OBJECT:
                bodyBuffer.writeString(((JsonObject) body).encode());
                break;
             case VertxConstants.TYPE_JSON_ARRAY:
                bodyBuffer.writeString(((JsonArray) body).encode());
                break;
             case VertxConstants.TYPE_REPLY_FAILURE:
                ReplyException except = (ReplyException) body;
                bodyBuffer.writeInt(except.failureType().toInt());
                bodyBuffer.writeInt(except.failureCode());
                bodyBuffer.writeString(except.getMessage());
                break;
             default:
                throw new IllegalArgumentException("Invalid body type: " + type);
          }
       }

       private int getMessageType(Message<?> message) {

          Object body = message.body();

          if (message instanceof PingMessage) {
             return VertxConstants.TYPE_PING;
          }
          else if (body instanceof Buffer) {
             return VertxConstants.TYPE_BUFFER;
          }
          else if (body instanceof Boolean) {
             return VertxConstants.TYPE_BOOLEAN;
          }
          else if (body instanceof byte[]) {
             return VertxConstants.TYPE_BYTEARRAY;
          }
          else if (body instanceof Byte) {
             return VertxConstants.TYPE_BYTE;
          }
          else if (body instanceof Character) {
             return VertxConstants.TYPE_CHARACTER;
          }
          else if (body instanceof Double) {
             return VertxConstants.TYPE_DOUBLE;
          }
          else if (body instanceof Float) {
             return VertxConstants.TYPE_FLOAT;
          }
          else if (body instanceof Integer) {
             return VertxConstants.TYPE_INT;
          }
          else if (body instanceof Long) {
             return VertxConstants.TYPE_LONG;
          }
          else if (body instanceof Short) {
             return VertxConstants.TYPE_SHORT;
          }
          else if (body instanceof String) {
             return VertxConstants.TYPE_STRING;
          }
          else if (body instanceof JsonArray) {
             return VertxConstants.TYPE_JSON_ARRAY;
          }
          else if (body instanceof JsonObject) {
             return VertxConstants.TYPE_JSON_OBJECT;
          }
          else if (body instanceof ReplyException) {
             return VertxConstants.TYPE_REPLY_FAILURE;
          }
          throw new IllegalArgumentException("Type not supported: " + message);
       }

    }

    @Override
    public String toString() {
       return "[IncomingVertxEventHandler(" + connectorName + "), queueName: " + queueName + " host: " + host + " port: " + port + " vertxAddress: " + vertxAddress + "]";
    }
 }
	/*
	* 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.activemq.artemis.integration.vertx;

	import java.util.Map;

	import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
	import org.apache.activemq.artemis.api.core.SimpleString;
	import org.apache.activemq.artemis.core.persistence.StorageManager;
	import org.apache.activemq.artemis.core.postoffice.Binding;
	import org.apache.activemq.artemis.core.postoffice.PostOffice;
	import org.apache.activemq.artemis.core.server.ConnectorService;
	import org.apache.activemq.artemis.core.server.ServerMessage;
	import org.apache.activemq.artemis.core.server.impl.ServerMessageImpl;
	import org.apache.activemq.artemis.utils.ConfigurationHelper;
	import org.vertx.java.core.Handler;
	import org.vertx.java.core.buffer.Buffer;
	import org.vertx.java.core.eventbus.EventBus;
	import org.vertx.java.core.eventbus.Message;
	import org.vertx.java.core.eventbus.ReplyException;
	import org.vertx.java.core.eventbus.impl.PingMessage;
	import org.vertx.java.core.json.JsonArray;
	import org.vertx.java.core.json.JsonObject;
	import org.vertx.java.platform.PlatformLocator;
	import org.vertx.java.platform.PlatformManager;
	import org.vertx.java.spi.cluster.impl.hazelcast.HazelcastClusterManagerFactory;

	class IncomingVertxEventHandler implements ConnectorService {

	private final String connectorName;

	private final String queueName;

	private final int port;

	private final String host;

	private final int quorumSize;

	private final String haGroup;

	private final String vertxAddress;

	private EventBus eventBus;

	private PlatformManager platformManager;

	private EventHandler handler;

	private final StorageManager storageManager;

	private final PostOffice postOffice;

	private boolean isStarted = false;

	IncomingVertxEventHandler(String connectorName,
	Map<String, Object> configuration,
	StorageManager storageManager,
	PostOffice postOffice) {
	this.connectorName = connectorName;
	this.queueName = ConfigurationHelper.getStringProperty(VertxConstants.QUEUE_NAME, null, configuration);

	this.port = ConfigurationHelper.getIntProperty(VertxConstants.PORT, 0, configuration);
	this.host = ConfigurationHelper.getStringProperty(VertxConstants.HOST, "localhost", configuration);
	this.quorumSize = ConfigurationHelper.getIntProperty(VertxConstants.VERTX_QUORUM_SIZE, -1, configuration);
	this.haGroup = ConfigurationHelper.getStringProperty(VertxConstants.VERTX_HA_GROUP, "activemq", configuration);
	this.vertxAddress = ConfigurationHelper.getStringProperty(VertxConstants.VERTX_ADDRESS, "org.apache.activemq", configuration);

	this.storageManager = storageManager;
	this.postOffice = postOffice;
	}

	@Override
	public void start() throws Exception {
	if (this.isStarted) {
	return;
	}
	System.setProperty("vertx.clusterManagerFactory", HazelcastClusterManagerFactory.class.getName());
	if (quorumSize != -1) {
	platformManager = PlatformLocator.factory.createPlatformManager(port, host, quorumSize, haGroup);
	}
	else {
	platformManager = PlatformLocator.factory.createPlatformManager(port, host);
	}

	eventBus = platformManager.vertx().eventBus();

	Binding b = postOffice.getBinding(new SimpleString(queueName));
	if (b == null) {
	throw new Exception(connectorName + ": queue " + queueName + " not found");
	}

	handler = new EventHandler();
	eventBus.registerHandler(vertxAddress, handler);

	isStarted = true;
	ActiveMQVertxLogger.LOGGER.debug(connectorName + ": started");
	}

	@Override
	public void stop() throws Exception {
	if (!isStarted) {
	return;
	}
	eventBus.unregisterHandler(vertxAddress, handler);
	platformManager.stop();
	System.clearProperty("vertx.clusterManagerFactory");
	isStarted = false;
	ActiveMQVertxLogger.LOGGER.debug(connectorName + ": stopped");
	}

	@Override
	public boolean isStarted() {
	return isStarted;
	}

	@Override
	public String getName() {
	return connectorName;
	}

	private class EventHandler implements Handler<Message<?>> {

	@Override
	public void handle(Message<?> message) {
	ServerMessage msg = new ServerMessageImpl(storageManager.generateID(), VertxConstants.INITIAL_MESSAGE_BUFFER_SIZE);
	msg.setAddress(new SimpleString(queueName));
	msg.setDurable(true);
	msg.encodeMessageIDToBuffer();

	String replyAddress = message.replyAddress();
	if (replyAddress != null) {
	msg.putStringProperty(VertxConstants.VERTX_MESSAGE_REPLYADDRESS, replyAddress);
	}

	// it'd be better that Message expose its type information
	int type = getMessageType(message);

	msg.putIntProperty(VertxConstants.VERTX_MESSAGE_TYPE, type);

	manualEncodeVertxMessageBody(msg.getBodyBuffer(), message.body(), type);

	try {
	postOffice.route(msg, null, false);
	}
	catch (Exception e) {
	ActiveMQVertxLogger.LOGGER.error("failed to route msg " + msg, e);
	}
	}

	private void manualEncodeVertxMessageBody(ActiveMQBuffer bodyBuffer, Object body, int type) {
	switch (type) {
	case VertxConstants.TYPE_BOOLEAN:
	bodyBuffer.writeBoolean(((Boolean) body));
	break;
	case VertxConstants.TYPE_BUFFER:
	Buffer buff = (Buffer) body;
	int len = buff.length();
	bodyBuffer.writeInt(len);
	bodyBuffer.writeBytes(((Buffer) body).getBytes());
	break;
	case VertxConstants.TYPE_BYTEARRAY:
	byte[] bytes = (byte[]) body;
	bodyBuffer.writeInt(bytes.length);
	bodyBuffer.writeBytes(bytes);
	break;
	case VertxConstants.TYPE_BYTE:
	bodyBuffer.writeByte((byte) body);
	break;
	case VertxConstants.TYPE_CHARACTER:
	bodyBuffer.writeChar((Character) body);
	break;
	case VertxConstants.TYPE_DOUBLE:
	bodyBuffer.writeDouble((double) body);
	break;
	case VertxConstants.TYPE_FLOAT:
	bodyBuffer.writeFloat((Float) body);
	break;
	case VertxConstants.TYPE_INT:
	bodyBuffer.writeInt((Integer) body);
	break;
	case VertxConstants.TYPE_LONG:
	bodyBuffer.writeLong((Long) body);
	break;
	case VertxConstants.TYPE_SHORT:
	bodyBuffer.writeShort((Short) body);
	break;
	case VertxConstants.TYPE_STRING:
	case VertxConstants.TYPE_PING:
	bodyBuffer.writeString((String) body);
	break;
	case VertxConstants.TYPE_JSON_OBJECT:
	bodyBuffer.writeString(((JsonObject) body).encode());
	break;
	case VertxConstants.TYPE_JSON_ARRAY:
	bodyBuffer.writeString(((JsonArray) body).encode());
	break;
	case VertxConstants.TYPE_REPLY_FAILURE:
	ReplyException except = (ReplyException) body;
	bodyBuffer.writeInt(except.failureType().toInt());
	bodyBuffer.writeInt(except.failureCode());
	bodyBuffer.writeString(except.getMessage());
	break;
	default:
	throw new IllegalArgumentException("Invalid body type: " + type);
	}
	}

	private int getMessageType(Message<?> message) {

	Object body = message.body();

	if (message instanceof PingMessage) {
	return VertxConstants.TYPE_PING;
	}
	else if (body instanceof Buffer) {
	return VertxConstants.TYPE_BUFFER;
	}
	else if (body instanceof Boolean) {
	return VertxConstants.TYPE_BOOLEAN;
	}
	else if (body instanceof byte[]) {
	return VertxConstants.TYPE_BYTEARRAY;
	}
	else if (body instanceof Byte) {
	return VertxConstants.TYPE_BYTE;
	}
	else if (body instanceof Character) {
	return VertxConstants.TYPE_CHARACTER;
	}
	else if (body instanceof Double) {
	return VertxConstants.TYPE_DOUBLE;
	}
	else if (body instanceof Float) {
	return VertxConstants.TYPE_FLOAT;
	}
	else if (body instanceof Integer) {
	return VertxConstants.TYPE_INT;
	}
	else if (body instanceof Long) {
	return VertxConstants.TYPE_LONG;
	}
	else if (body instanceof Short) {
	return VertxConstants.TYPE_SHORT;
	}
	else if (body instanceof String) {
	return VertxConstants.TYPE_STRING;
	}
	else if (body instanceof JsonArray) {
	return VertxConstants.TYPE_JSON_ARRAY;
	}
	else if (body instanceof JsonObject) {
	return VertxConstants.TYPE_JSON_OBJECT;
	}
	else if (body instanceof ReplyException) {
	return VertxConstants.TYPE_REPLY_FAILURE;
	}
	throw new IllegalArgumentException("Type not supported: " + message);
	}

	}

	@Override
	public String toString() {
	return "[IncomingVertxEventHandler(" + connectorName + "), queueName: " + queueName + " host: " + host + " port: " + port + " vertxAddress: " + vertxAddress + "]";
	}
	}