
package org.apache.activemq.artemis.core.protocol.proton.converter;

import org.apache.qpid.proton.jms.EncodedMessage;
import org.apache.qpid.proton.jms.InboundTransformer;
import org.apache.qpid.proton.jms.JMSMappingInboundTransformer;
import org.apache.qpid.proton.jms.JMSMappingOutboundTransformer;
import org.apache.activemq.artemis.core.protocol.proton.converter.jms.ServerJMSMessage;
import org.apache.activemq.artemis.core.server.ServerMessage;
import org.apache.activemq.artemis.spi.core.protocol.MessageConverter;
import org.apache.activemq.artemis.utils.IDGenerator;

public class ProtonMessageConverter implements MessageConverter
{


   ActiveMQJMSVendor activeMQJMSVendor;

   public ProtonMessageConverter(IDGenerator idGenerator)
   {
      activeMQJMSVendor = new ActiveMQJMSVendor(idGenerator);
      inboundTransformer = new JMSMappingInboundTransformer(activeMQJMSVendor);
      outboundTransformer = new JMSMappingOutboundTransformer(activeMQJMSVendor);
   }

   private final InboundTransformer inboundTransformer;
   private final JMSMappingOutboundTransformer outboundTransformer;

   @Override
   public ServerMessage inbound(Object messageSource) throws Exception
   {
      ServerJMSMessage jmsMessage = inboundJMSType((EncodedMessage) messageSource);

      return (ServerMessage)jmsMessage.getInnerMessage();
   }

   /**
    * Just create the JMS Part of the inbound (for testing)
    * @param messageSource
    * @return
    * @throws Exception
    */
   public ServerJMSMessage inboundJMSType(EncodedMessage messageSource) throws Exception
   {
      EncodedMessage encodedMessageSource = messageSource;
      ServerJMSMessage transformedMessage = (ServerJMSMessage)inboundTransformer.transform(encodedMessageSource);

      transformedMessage.encode();

      return transformedMessage;
   }


   @Override
   public Object outbound(ServerMessage messageOutbound, int deliveryCount) throws Exception
   {
      ServerJMSMessage jmsMessage = activeMQJMSVendor.wrapMessage(messageOutbound.getType(), messageOutbound, deliveryCount);
      jmsMessage.decode();

      return outboundTransformer.convert(jmsMessage);
   }
}
