package org.yamcs.tctm;

import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
import org.apache.activemq.artemis.api.core.client.MessageHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yamcs.ConfigurationException;
import org.yamcs.YamcsServer;
import org.yamcs.api.YamcsApiException;
import org.yamcs.api.artemis.Protocol;
import org.yamcs.api.artemis.YamcsClient;
import org.yamcs.api.artemis.YamcsSession;
import org.yamcs.archive.PacketWithTime;
import org.yamcs.artemis.AbstractArtemisTranslatorService;

import com.google.common.util.concurrent.AbstractService;

import org.yamcs.protobuf.Yamcs.TmPacketData;
import org.yamcs.time.TimeService;


/**
 * receives data from Artemis ActiveMQ and publishes it into a yamcs stream
 * 
 * @author nm
 *
 */
public class ArtemisTmDataLink extends  AbstractService implements TmPacketDataLink, MessageHandler {
    protected volatile long packetcount = 0;
    protected volatile boolean disabled=false;
   
    protected Logger log=LoggerFactory.getLogger(this.getClass().getName());
    private TmSink tmSink;
    YamcsSession yamcsSession; 
    final private YamcsClient msgClient;
    final TimeService timeService;

    public ArtemisTmDataLink(String instance, String name, String hornetAddress) throws ConfigurationException  {
        SimpleString queue = new SimpleString(hornetAddress+"-ActiveMQTmProvider");

        try {
            yamcsSession=YamcsSession.newBuilder().build();
            msgClient=yamcsSession.newClientBuilder().setDataProducer(false).setDataConsumer(new SimpleString(hornetAddress), queue).
                    setFilter(new SimpleString(AbstractArtemisTranslatorService.UNIQUEID_HDR_NAME+"<>"+AbstractArtemisTranslatorService.UNIQUEID)).
                    build();

        } catch (Exception e) {
            throw new ConfigurationException(e.getMessage(),e);
        }
        timeService = YamcsServer.getTimeService(instance);
    }


    @Override
    public void setTmSink(TmSink tmProcessor) {
        this.tmSink=tmProcessor;
    }

    @Override
    public String getLinkStatus() {
        if (disabled) {
            return "DISABLED";
        } else {
            return "OK";
        }
    }

    @Override
    public void disable() {
        disabled=true;
    }

    @Override
    public void enable() {
        disabled=false;
    }

    @Override
    public boolean isDisabled() {
        return disabled;
    }

    @Override
    public String getDetailedStatus() {
        if(disabled) {
            return "DISABLED";
        } else {
            return "OK";
        }
    }

    @Override
    public long getDataCount() {
        return packetcount;
    }


    @Override
    public void onMessage(ClientMessage msg) {
        if(disabled) return;
        try {
            TmPacketData tm=(TmPacketData)Protocol.decode(msg, TmPacketData.newBuilder());
            packetcount++;
            //System.out.println("mark 1: message received: "+msg);
            PacketWithTime pwt =  new PacketWithTime(timeService.getMissionTime(), tm.getGenerationTime(), tm.getPacket().toByteArray());
            tmSink.processPacket(pwt);
        } catch(YamcsApiException e){
            log.warn( "{} for message: {}", e.getMessage(), msg);
        }
    }

    @Override
    protected void doStart() {
        try {
            msgClient.dataConsumer.setMessageHandler(this);
            notifyStarted();
        } catch (ActiveMQException e) {
            log.error("Failed to set message handler");
            notifyFailed(e);
        }
    }

    @Override
    protected void doStop() {
        try {
            msgClient.close();
            notifyStopped();
        } catch (ActiveMQException e) {
            log.error("Got exception when quiting:", e);
            notifyFailed(e);
        }
    }

}


@@ -1250,22 +1250,16 @@
 n; %0A    
-final 
 private 
@@ -1315,16 +1315,54 @@
 Service;
+%0A    final String artemisAddress;%0A    
 %0A%0A    pu
@@ -1421,22 +1421,23 @@
  String 
-hornet
+artemis
 Address)
@@ -1482,576 +1482,45 @@
     
-SimpleString queue = new SimpleString(hornetAddress+%22-ActiveMQTmProvider%22);%0A%0A        try %7B%0A            yamcsSession=YamcsSession.newBuilder().build();%0A            msgClient=yamcsSession.newClientBuilder().setDataProducer(false).setDataConsumer(new SimpleString(hornetAddress), queue).%0A                    setFilter(new SimpleString(AbstractArtemisTranslatorService.UNIQUEID_HDR_NAME+%22%3C%3E%22+AbstractArtemisTranslatorService.UNIQUEID)).%0A                    build();%0A%0A        %7D catch (Exception e) %7B%0A            throw new ConfigurationException(e.getMessage(),e);%0A        %7D
+this.artemisAddress = artemisAddress;
 %0A   
@@ -2963,32 +2963,514 @@
 %7B%0A        try %7B%0A
+            SimpleString queue = new SimpleString(artemisAddress+%22-ActiveMQTmProvider%22);%0A                yamcsSession = YamcsSession.newBuilder().build();%0A                msgClient = yamcsSession.newClientBuilder().setDataProducer(false).setDataConsumer(new SimpleString(artemisAddress), queue).%0A                        setFilter(new SimpleString(AbstractArtemisTranslatorService.UNIQUEID_HDR_NAME+%22%3C%3E%22+AbstractArtemisTranslatorService.UNIQUEID)).%0A                        build();%0A%0A
             msgC
@@ -3569,32 +3569,52 @@
 tiveMQException 
+%7C YamcsApiException 
 e) %7B%0A           
@@ -3643,23 +3643,26 @@
 set 
-message handler
+connect to artemis
 %22);%0A

connect to artemis only in startUp because it is not yet started due to the change in start order