/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2017 - 2024 ScalAgent Distributed Technologies
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA.
 *
 * Initial developer(s): ScalAgent Distributed Technologies
 * Contributor(s): 
 */
package com.scalagent.joram.mom.dest.rest;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.Constructor;
import java.net.URI;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import jakarta.ws.rs.client.Client;
import jakarta.ws.rs.client.ClientBuilder;
import jakarta.ws.rs.client.Entity;
import jakarta.ws.rs.client.Invocation.Builder;
import jakarta.ws.rs.client.WebTarget;
import jakarta.ws.rs.core.Form;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.UriBuilder;
import jakarta.ws.rs.ext.RuntimeDelegate;

import org.objectweb.joram.mom.dest.AcquisitionHandler;
import org.objectweb.joram.mom.dest.ReliableTransmitter;
import org.objectweb.joram.shared.DestinationConstants;
import org.objectweb.joram.shared.excepts.MessageValueException;
import org.objectweb.joram.shared.messages.ConversionHelper;
import org.objectweb.joram.shared.messages.Message;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

import fr.dyade.aaa.common.Debug;

/**
 * Simple acquisition handler for the REST acquisition bridge.
 * 
 * A Rest/JMS acquisition request is triggered each time a new message is received. If this
 * message contains properties, the module is reinitialized with the incoming properties.
 */
public final class RESTAcquisition implements AcquisitionHandler {
  private static final Logger logger = Debug.getLogger(RESTAcquisition.class.getName());
  
  static {
    System.setProperty(RuntimeDelegate.JAXRS_RUNTIME_DELEGATE_PROPERTY, "org.glassfish.jersey.internal.RuntimeDelegateImpl");
    System.setProperty(ClientBuilder.JAXRS_DEFAULT_CLIENT_BUILDER_PROPERTY, "org.glassfish.jersey.client.JerseyClientBuilder");
  }
  
  private Properties properties = null;

  private String hostName = "localhost";
  private int port = 8989;

  // If true, use the API with authentication parameters in URL.
  private boolean useOldAPI = false;

  private int connectTimeout = 5000;
  // Be careful, we need to add timeout for the client configuration.
  private int readTimeout = 5000;
  
  private String userName = null;
  private String password = null;

  // Normally each consumer resource need to be explicitly closed, this parameter allows to set the idle time
  // in seconds in which the consumer context will be closed if idle.
  private String idleTimeout = "60";
  // Timeout for waiting for a message.
  private long timeout = 0; //timeout = 0 => NoWait
  private boolean persistent = false;
  private int nbMaxMsg = 100;
  private boolean mediaTypeJson = true;//default true "application/json"
  
  private String destName;
  
  private void init(Properties properties) {
    this.properties = properties;

    destName = properties.getProperty(DestinationConstants.DESTINATION_NAME_PROP);
    if (destName == null) {
      throw new IllegalArgumentException("Missing Destination JNDI name.");
    } else {
      try {
        destName = URLEncoder.encode(destName, "UTF-8");
      } catch (UnsupportedEncodingException exc) {
        // Should never happened
      }
    }
    
    if (properties.containsKey(DestinationConstants.REST_HOST_PROP)) {
      hostName = properties.getProperty(DestinationConstants.REST_HOST_PROP);
    }
    if (properties.containsKey(DestinationConstants.REST_PORT_PROP)) {
      try {
        port = Integer.parseInt(properties.getProperty(DestinationConstants.REST_PORT_PROP));
      } catch (NumberFormatException nfe) {
        logger.log(BasicLevel.ERROR,
                   "Property " + DestinationConstants.REST_PORT_PROP + '=' + properties.getProperty(DestinationConstants.REST_PORT_PROP) +
                   " could not be parsed properly, use default value: " + port);
      }
    }
    
    if (properties.containsKey(DestinationConstants.REST_USE_OLD_API)) {
      useOldAPI = Boolean.parseBoolean(properties.getProperty(DestinationConstants.REST_USE_OLD_API));
      logger.log(BasicLevel.INFO, DestinationConstants.REST_USE_OLD_API + "=" + useOldAPI);
    }

    if (properties.containsKey(DestinationConstants.REST_USERNAME_PROP)) {
      userName = properties.getProperty(DestinationConstants.REST_USERNAME_PROP);
    }
    if (properties.containsKey(DestinationConstants.REST_PASSWORD_PROP)) {
      password = properties.getProperty(DestinationConstants.REST_PASSWORD_PROP);
    }
    
    if (properties.containsKey(DestinationConstants.TIMEOUT_PROP)) {
      try {
        timeout = Long.parseLong(properties.getProperty(DestinationConstants.TIMEOUT_PROP));
      } catch (NumberFormatException exc) { }
    }
    if (properties.containsKey(DestinationConstants.IDLETIMEOUT_PROP)) {
        idleTimeout = properties.getProperty(DestinationConstants.IDLETIMEOUT_PROP);
    } else {
      logger.log(BasicLevel.WARN,
                 "Missing property " + DestinationConstants.IDLETIMEOUT_PROP + ", use default value: " + idleTimeout);
    }
    
    if (properties.containsKey(DestinationConstants.NB_MAX_MSG_PROP)) {
      try {
        nbMaxMsg = Integer.parseInt(properties.getProperty(DestinationConstants.NB_MAX_MSG_PROP));
      } catch (NumberFormatException exc) { }
    }

    if (properties.containsKey(DestinationConstants.REST_CONNECT_TIMEOUT_PROP)) {
      try {
        connectTimeout = Integer.parseInt(properties.getProperty(DestinationConstants.REST_CONNECT_TIMEOUT_PROP));
      } catch (NumberFormatException exc) {
        logger.log(BasicLevel.ERROR,
            "Property " + DestinationConstants.REST_CONNECT_TIMEOUT_PROP + '=' + properties.getProperty(DestinationConstants.REST_CONNECT_TIMEOUT_PROP) + 
            " could not be parsed properly, use default value: " + connectTimeout);
      }
    } else {
      logger.log(BasicLevel.WARN,
          "Missing property " + DestinationConstants.REST_CONNECT_TIMEOUT_PROP + ", use default value: " + connectTimeout);
    }

    if (properties.containsKey(DestinationConstants.REST_READ_TIMEOUT_PROP)) {
      try {
        readTimeout = Integer.parseInt(properties.getProperty(DestinationConstants.REST_READ_TIMEOUT_PROP));
        readTimeout += timeout;
      } catch (NumberFormatException exc) {
        logger.log(BasicLevel.ERROR,
            "Property " + DestinationConstants.REST_READ_TIMEOUT_PROP + '=' + properties.getProperty(DestinationConstants.REST_READ_TIMEOUT_PROP) + 
            " could not be parsed properly, use default value: " + readTimeout);
      }
    } else {
      logger.log(BasicLevel.WARN,
          "Missing property " + DestinationConstants.REST_READ_TIMEOUT_PROP + ", use default value: " + readTimeout);
    }
    
    if (properties.containsKey(DestinationConstants.MEDIA_TYPE_JSON_PROP)) {
      mediaTypeJson = Boolean.parseBoolean(properties.getProperty(DestinationConstants.MEDIA_TYPE_JSON_PROP));
    }
    
    try {
      createConsumer();
    } catch (Exception exc) {
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.ERROR,
                   "RestAcquisitionc.createConsumer(): cannot creates consumer: " + destName, exc);

      
    }
  }
  
  private URI uriConsume;
  private URI uriCloseConsumer;
  
  private Client client;

  /**
   * This method is called either from init (implying initialization of the consumer), or retrieve if
   * the consumer is not initialized.
   * 
   * @throws Exception
   */
  private void createConsumer() throws Exception {
    // Close the consumer if it is already alive
    closeConsumer();

    Response response1 = null, response2 = null;
    try {
      URI base = UriBuilder.fromUri("http://" + hostName + ":" + port + "/joram/").build();
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG,
                   "RestAcquisition.createConsumer(), use base URI " + base);

      // Initializes URI
      uriConsume = null;
      uriCloseConsumer = null;

      // Initializes Rest client and target
      try {
        client = ClientBuilder.newBuilder()
            .connectTimeout(connectTimeout, TimeUnit.MILLISECONDS).readTimeout(readTimeout, TimeUnit.MILLISECONDS).newClient();
        for (Object key : properties.keySet()) {
          try {
            if (key instanceof String)
              client.property((String) key, properties.get(key));
          } catch (Exception e) {
            logger.log(BasicLevel.ERROR,
                       "RestAcquisition.createConsumer(): cannot set client property " + key + "=" + properties.get(key));
          }
        }
      } catch (Exception exc) {
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.ERROR,
                     "RestAcquisition.createConsumer(): cannot initialize Rest client", exc);
        else
          logger.log(BasicLevel.ERROR,
                     "RestAcquisition.createConsumer(): cannot initialize Rest client - " + exc.getMessage());
        throw exc;
      }

      // Get the destination
      try {
        response1 = client.target(base).path("jndi").path(destName).request().accept(MediaType.TEXT_PLAIN).head();
      } catch (Exception exc) {
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.ERROR,
                     "RestAcquisitionc.createConsumer(): cannot get destination " + destName, exc);
        else
          logger.log(BasicLevel.ERROR,
                     "RestAcquisition.createConsumer(): cannot get destination " + destName + " - " + exc.getMessage());
        throw exc;
      }

      if (response1.getStatus() == 201) {
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG,
                     "RestAcquisition.createConsumer(): get destination -> " + response1.getStatusInfo());
      } else {
        if (logger.isLoggable(BasicLevel.ERROR))
          logger.log(BasicLevel.ERROR,
                     "RestAcquisition.createConsumer(): cannot get destination " + destName + " -> " + response1.getStatusInfo());
        throw new Exception("Cannot get destination " + destName);
      }

      try {
        WebTarget target;
        if (useOldAPI) {
          URI uriCreateConsumer = response1.getLink("create-consumer").getUri();
          if (logger.isLoggable(BasicLevel.DEBUG))
            logger.log(BasicLevel.DEBUG,
                       "RESTAcquisition.createConsumer(): create-consumer = " + uriCreateConsumer);

          target = client.target(uriCreateConsumer);
        } else {
          target = client.target(base).path("jndi").path(destName).path(org.objectweb.joram.tools.rest.jms.JmsService.JMS_CREATE_CONS_FP);
        }
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG, "RestAcquisition.createConsumer(): create-consumer = " + target.getUri());

        // TODO (AF): We should fix name and client-id to be unique.
        target = target.queryParam("name", "cons-" + destName);
        target = target.queryParam("client-id", "id-" + destName);
        // Normally each consumer resource need to be explicitly closed, this parameter allows to set the idle time
        // in seconds in which the consumer context will be closed if idle.
        if (idleTimeout != null)  target = target.queryParam("idle-timeout", idleTimeout);

        Form auth = new Form();
        if (useOldAPI) {
          if (userName != null) target = target.queryParam("user", userName);
          if (password != null) target = target.queryParam("password", password);
        } else {
          if (userName != null) auth.param("user", userName);
          if (password != null) auth.param("password", password);
        }

        if (useOldAPI) {
          response2 = target.request().accept(MediaType.TEXT_PLAIN).post(null);
        } else {
          response2 = target.request().accept(MediaType.TEXT_PLAIN).post(Entity.entity(auth, MediaType.APPLICATION_FORM_URLENCODED));
        }
      } catch (Exception exc) {
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.ERROR,
                     "RestAcquisition.createConsumer(): cannot create consumer", exc);
        else
          logger.log(BasicLevel.ERROR,
                     "RestAcquisition.createConsumer(): cannot create consumer - " + exc.getMessage());
        throw exc;
      }

      if (response2.getStatus() == 201) {
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG,
                     "RESTAcquisition.createConsumer: uriCloseConsumer = " + uriCloseConsumer + ", uriConsume = " + uriConsume);

        uriCloseConsumer = response2.getLink("close-context").getUri();
        uriConsume = response2.getLink("receive-message").getUri();
      } else {
        if (logger.isLoggable(BasicLevel.ERROR))
          logger.log(BasicLevel.ERROR,
                     "RestAcquisition.createConsumer(): cannot create consumer -> " + response2.getStatusInfo());
        throw new Exception("Cannot create consumer");
      }
    } finally {
      if (response1 != null) response1.close();
      if (response2 != null) response2.close();
      
      if (uriCloseConsumer == null) {
        // the createConsumer has failed, closes the client if needed.
        if (client != null) client.close();
        client = null;
      }
    }
  }
  
  private void setMessageHeader(Map jsonMessageHeader, Message message) {
    if (jsonMessageHeader.containsKey("DeliveryMode"))
      try {
        message.persistent = "PERSISTENT".equals(jsonMessageHeader.get("DeliveryMode"));
      } catch (Exception e) { 
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "-- DeliveryMode = " + jsonMessageHeader.get("DeliveryMode"));
      }
    
    if (jsonMessageHeader.containsKey("Priority")) {
      try {
        message.priority = ((Double) jsonMessageHeader.get("Priority")).intValue();
      } catch (Exception e) { 
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "-- Priority = " + jsonMessageHeader.get("Priority"));
      }
    }
    
    if (jsonMessageHeader.containsKey("Redelivered")) {
      try {
        message.redelivered = (boolean) jsonMessageHeader.get("Redelivered");
      } catch (Exception e) { 
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "-- Redelivered = " + jsonMessageHeader.get("Redelivered"));
      }
    }

    if (jsonMessageHeader.containsKey("Timestamp")) {
      try {
        message.timestamp = ((Double) jsonMessageHeader.get("Timestamp")).longValue();
      } catch (Exception e) { 
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "-- Timestamp = " + jsonMessageHeader.get("Timestamp"));
      }
    }
    
    if (jsonMessageHeader.containsKey("Expiration")) {
      try {
        message.expiration = ((Double) jsonMessageHeader.get("Expiration")).longValue();
      } catch (Exception e) { 
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "-- Expiration = " + jsonMessageHeader.get("Expiration"));
      }
    }
    
    if (jsonMessageHeader.containsKey("CorrelationID"))
      try {
        message.correlationId = (String) jsonMessageHeader.get("CorrelationID");
      } catch (Exception e) {
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "-- CorrelationID = " + jsonMessageHeader.get("CorrelationID"));
      }
    
    if (jsonMessageHeader.containsKey("CorrelationIDAsBytes")) {
      message.setJMSCorrelationIDAsBytes((byte[]) jsonMessageHeader.get("CorrelationIDAsBytes"));
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "-- CorrelationIDAsBytes = " + jsonMessageHeader.get("CorrelationIDAsBytes"));
    }
    
    if (jsonMessageHeader.containsKey("Destination")) {
      try {
        Map dest = (Map) jsonMessageHeader.get("Destination");
        String id = (String) dest.get("agentId");
        String name = (String) dest.get("adminName");
        byte type = ((Double) dest.get("type")).byteValue();
        message.setDestination(id, name, type);
      } catch (Exception e) {
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "-- Destination = " + jsonMessageHeader.get("Destination"));
      }
    }
    
    if (jsonMessageHeader.containsKey("MessageID"))
      try {
        message.id = (String) jsonMessageHeader.get("MessageID");
      } catch (Exception e) {
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "-- MessageID = " + jsonMessageHeader.get("MessageID"));
      }
      
    if (jsonMessageHeader.containsKey("ReplyTo"))
      try {
        message.replyToId = (String) jsonMessageHeader.get("ReplyTo");
      } catch (Exception e) {
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "-- ReplyTo = " + jsonMessageHeader.get("ReplyTo"));
      }
    
    if (jsonMessageHeader.containsKey("Type")) {
      try {
        message.jmsType = (String) jsonMessageHeader.get("Type");
      } catch (Exception e) { 
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, "-- Type = " + jsonMessageHeader.get("Type"));
      }
    }
  }
  
  private HashMap getMapMessage(Map<String, Object> jsonMap) throws Exception {
    if (jsonMap == null)
      return null;

    HashMap map = new HashMap<>();
    // parse the json map
    for (String key : jsonMap.keySet()) {
      Object value = jsonMap.get(key);
      if (value instanceof ArrayList) {
        ArrayList<String> array =(ArrayList<String>) value; 
        try {
          if (array.size() == 2) {
            String className = array.get(1);
            if (Character.class.getName().equals(className)) {
              value =  array.get(0).charAt(0);
            } else if (byte[].class.getName().equals(className)) {
             value = array.get(0).getBytes("UTF-8");
            } else {
              Constructor<?> constructor = Class.forName(className).getConstructor(String.class);
              value = constructor.newInstance(array.get(0));
            }
            map.put(key, value);
          }
        } catch (Exception e) {
          logger.log(BasicLevel.ERROR, "getMapMessage: ignore map entry " + key + ", " + value + " : " + e.getMessage());
          continue;
        }
      }
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG,
                   "getMapMessage: " + key + ", value = " + value + ", " + value.getClass().getSimpleName());
    }
    return map;
  }
  
  @Override
  public void retrieve(ReliableTransmitter transmitter) throws Exception {
    if (uriCloseConsumer == null)
      createConsumer();
    
    if (uriConsume == null)
      return;
    
    try {
      ArrayList<Message> messages = new ArrayList<Message>();
      while (messages.size() < nbMaxMsg) {
        //timeout = 0 => NoWait
        Builder builder = null;
        Message message = null;
        if (!mediaTypeJson) {
          //TEXT
          builder = client.target(uriConsume)
              .queryParam("timeout", timeout)
              .request().accept(MediaType.TEXT_PLAIN);
          Response response = builder.get();
          String msg = response.readEntity(String.class);
          if (200 != response.getStatus() || msg == null || msg.isEmpty())
            break;
          message = new Message();
          message.type = Message.TEXT;
          message.setText(msg);
        } else {
          //JSON
          builder = client.target(uriConsume)
              .queryParam("timeout", timeout)
              .request().accept(MediaType.APPLICATION_JSON);

          Response response = builder.get();

          //        if (logger.isLoggable(BasicLevel.DEBUG))
          //          logger.log(BasicLevel.DEBUG, "RESTAcquisition.retrieve response = " + response);
          String json = response.readEntity(String.class);

          if (200 != response.getStatus() || json == null || json.isEmpty() || "null".equals(json))
            break;

          try {
            Gson gson = new GsonBuilder().create();
            HashMap<String, Object> jsonMsg = gson.fromJson(json, HashMap.class);
            if (jsonMsg == null) {
              String text = gson.fromJson(json, String.class);
              if (text != null) {
                message = new Message();
                message.type = Message.TEXT;
                message.setText(text);
              }
            } else {
              if (logger.isLoggable(BasicLevel.DEBUG))
                logger.log(BasicLevel.DEBUG, "RESTAcquisition.retrieve msg = " + jsonMsg);

              message = new Message();
              //get Properties
              Map jmsProperties = (Map) jsonMsg.get("properties");
              if (jmsProperties != null && jmsProperties.size() > 0) {
                Set<Map.Entry> entrySet = jmsProperties.entrySet();
                for (Map.Entry entry : entrySet) {
                  String key = (String) entry.getKey();
                  ArrayList value = (ArrayList) entry.getValue();
                  String propType = (String) value.get(1);
                  Object propValue = null;
                  switch (propType) {
                  case "java.lang.Boolean":
                    propValue = Boolean.parseBoolean((String)value.get(0));
                    break;
                  case "java.lang.Byte":
                    propValue = Byte.parseByte((String)value.get(0));
                    break;
                  case "java.lang.Short":
                    propValue = Short.parseShort((String)value.get(0));
                    break;
                  case "java.lang.Integer":
                    propValue = Integer.parseInt((String)value.get(0));
                    break;
                  case "java.lang.Long":
                    propValue = Long.parseLong((String)value.get(0));
                    break;
                  case "java.lang.Float":
                    propValue = Float.parseFloat((String)value.get(0));
                    break;
                  case "java.lang.Double":
                    propValue = Double.parseDouble((String)value.get(0));
                    break;
                  case "java.lang.String":
                    propValue = (String)value.get(0);
                    break;

                  default:
                    if (logger.isLoggable(BasicLevel.WARN))
                      logger.log(BasicLevel.WARN, "RESTAcquisition.retrieve property type not supported: " + propType);
                    break;
                  }
                  if (key != null && propValue != null)
                    message.setProperty(key, propValue);
                }
              }

              //Get Header
              Map jmsHeader = (Map) jsonMsg.get("header");
              if (jmsHeader != null && jmsHeader.size() > 0) {
                setMessageHeader(jmsHeader, message);
              }

              //Get body
              String type = (String) jsonMsg.get("type");
              message.jmsType = type;
              switch (type) {
              case "TextMessage": {
                String jmsBody =  (String) jsonMsg.get("body");
                try {
                  message.type = Message.TEXT;
                  message.setText(jmsBody);
                } catch (IOException exc) {
                  MessageFormatException jExc = new MessageFormatException("The message body could not be serialized.");
                  jExc.setLinkedException(exc);
                  throw jExc;
                }
              } break;

              case "MapMessage": {
                Map jsonBody =  (Map) jsonMsg.get("body");
                try {
                  HashMap map = getMapMessage(jsonBody);
                  message.type = Message.MAP;
                  message.setMap(map);
                } catch (IOException exc) {
                  MessageFormatException jExc = new MessageFormatException("The message body could not be serialized.");
                  jExc.setLinkedException(exc);
                  throw jExc;
                }
              } break;

              case "BytesMessage": {
                ArrayList jmsBody =  (ArrayList) jsonMsg.get("body");
                message.type = Message.BYTES;
                byte[] bytes = new byte[jmsBody.size()];
                for (int i = 0; i < jmsBody.size(); i++) {
                  Object value = jmsBody.get(i);
                  bytes[i] = ((Number) value).byteValue();
                }
                message.setBody(bytes);
              } break;

              default:
                logger.log(BasicLevel.ERROR, "RESTAcquisition.retrieve type = " + type + " not supported.");
                break;
              }
            }
          } catch (Exception e) {
            if (logger.isLoggable(BasicLevel.DEBUG))
              logger.log(BasicLevel.WARN, "RESTAcquisition.retrieve json = " + json, e);
            else
              logger.log(BasicLevel.WARN, "RESTAcquisition.retrieve json = " + json + "\n\t" + e.getMessage());
            message = null;
          }
        }
        
        // TODO (AF): Get the new uriConsume URI ?
        
        if (message != null)
          messages.add(message);
      }

      if (messages.size() > 0)
        transmitter.transmit(messages, persistent);
    } catch (Exception e) {
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.WARN, "RESTAcquisition.retrieve", e);
      else
        logger.log(BasicLevel.WARN, "RESTAcquisition.retrieve:" + e.getMessage());
    }
  }

  @Override
  public void setProperties(Properties properties) {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "RESTAcquisition.setProperties properties = " + properties);
    
    if (properties.containsKey(DestinationConstants.ACQUISITION_PERSISTENT)) {
      try {
        persistent = ConversionHelper.toBoolean(properties.get(DestinationConstants.ACQUISITION_PERSISTENT));
      } catch (MessageValueException e) {
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.WARN, "RESTAcquisition.setProperties", e);
        else
          logger.log(BasicLevel.WARN, "RESTAcquisition.setProperties: " + e.getMessage());
      }
    }
    
    init(properties);
  }

  /**
   * Close the consumer.
   */
  private void closeConsumer() {
    Response response = null;
    try {
      if (uriCloseConsumer == null) return;

      response = client.target(uriCloseConsumer).request().accept(MediaType.TEXT_PLAIN).delete();
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, "RESTAcquisition.closeConsumer: " + uriCloseConsumer + " -> " + response.getStatusInfo());
    } finally {
      if (response != null) response.close();
      uriCloseConsumer = null;
      // A new client is created in createConsumer
      if (client != null) client.close();
      client = null;
    }
  }
  
  /**
   * Closes the Acquisition module and all associated resources.
   * Called by generic behavior during the destination finalization.
   */
  @Override
  public void close() {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "RestDistribution.close()");
    closeConsumer();
  }
}
