Communicator.java

package eu.chargetime.ocpp;

/*
ChargeTime.eu - Java-OCA-OCPP
Copyright (C) 2015-2016 Thomas Volden <tv@chargetime.eu>
Copyright (C) 2022 Emil Melar <emil@iconsultable.no>

MIT License

Copyright (C) 2016-2018 Thomas Volden

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/

import eu.chargetime.ocpp.feature.Feature;
import eu.chargetime.ocpp.model.*;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Abstract class. Handles basic communication: Pack and send messages. Receive and unpack messages.
 *
 * <p>Requires a {@link Transmitter} to send and receive messages. Must be overloaded to implement a
 * specific format.
 */
public abstract class Communicator {
  private static final Logger logger = LoggerFactory.getLogger(Communicator.class);

  private final ConcurrentLinkedDeque<Object> transactionQueue;
  private RetryRunner retryRunner;
  protected Radio radio;
  private CommunicatorEvents events;
  private final AtomicBoolean failedFlag = new AtomicBoolean(false);

  /**
   * Convert a formatted string into a {@link Request}/{@link Confirmation}. This is useful for call
   * results, where the confirmation type isn't given.
   *
   * @param payload the raw formatted payload.
   * @param type the expected return type.
   * @return the unpacked payload.
   * @param <T> the type of the unpacked payload.
   * @throws Exception error occurred while converting.
   */
  public abstract <T> T unpackPayload(Object payload, Class<T> type) throws Exception;

  /**
   * Convert a {@link Request}/{@link Confirmation} into a formatted string.
   *
   * @param payload the payload model.
   * @return the payload in the form of a formatted string.
   */
  public abstract Object packPayload(Object payload);

  /**
   * Create a call result envelope to transmit.
   *
   * @param uniqueId the id the receiver expects.
   * @param action action name of the {@link Feature}.
   * @param payload packed payload.
   * @return a fully packed message ready to send.
   */
  protected abstract Object makeCallResult(String uniqueId, String action, Object payload);

  /**
   * Create a call envelope to transmit to the server.
   *
   * @param uniqueId the id the receiver must reply with.
   * @param action action name of the {@link Feature}.
   * @param payload packed payload.
   * @return a fully packed message ready to send.
   */
  protected abstract Object makeCall(String uniqueId, String action, Object payload);

  /**
   * Create a call error envelope to transmit.
   *
   * @param uniqueId the id the receiver expects.
   * @param action action name of the {@link Feature}.
   * @param errorCode an OCPP error code.
   * @param errorDescription an associated error description.
   * @return a fully packed message ready to send.
   */
  protected abstract Object makeCallError(
      String uniqueId, String action, String errorCode, String errorDescription);

  /**
   * Create a call result error envelope to transmit.
   *
   * @param uniqueId the id the receiver expects.
   * @param action action name of the {@link Feature}.
   * @param errorCode an OCPP error code.
   * @param errorDescription an associated error description.
   * @return a fully packed message ready to send.
   */
  protected abstract Object makeCallResultError(
      String uniqueId, String action, String errorCode, String errorDescription);

  /**
   * Create a send envelope to transmit to the server.
   *
   * @param uniqueId the id of the message.
   * @param action action name of the {@link Feature}.
   * @param payload packed payload.
   * @return a fully packed message ready to send.
   */
  protected abstract Object makeSend(String uniqueId, String action, Object payload);

  /**
   * Identify an incoming call and parse it into one of the following: {@link CallMessage} a
   * request. {@link CallResultMessage} a response.
   *
   * @param message the raw message
   * @return CallMessage or {@link CallResultMessage}
   */
  protected abstract Message parse(Object message);

  /**
   * Handle required injections.
   *
   * @param transmitter Injected {@link Transmitter}
   */
  public Communicator(Radio transmitter) {
    this(transmitter, true);
  }

  /**
   * Handle required injections.
   *
   * @param transmitter Injected {@link Transmitter}
   * @param enableTransactionQueue flag to enable/disable the transaction queue and associated
   *     processing
   */
  public Communicator(Radio transmitter, boolean enableTransactionQueue) {
    this.radio = transmitter;
    this.transactionQueue = enableTransactionQueue ? new ConcurrentLinkedDeque<>() : null;
    this.retryRunner = enableTransactionQueue ? new RetryRunner() : null;
  }

  /**
   * Use the injected {@link Transmitter} to connect to server.
   *
   * @param uri the url and port of the server.
   * @param events handler for call back events.
   */
  public void connect(String uri, CommunicatorEvents events) {
    this.events = events;
    if (radio instanceof Transmitter) ((Transmitter) radio).connect(uri, new EventHandler(events));
  }

  /**
   * Use the injected {@link Transmitter} to accept a client.
   *
   * @param events handler for call back events.
   */
  public void accept(CommunicatorEvents events) {
    this.events = events;
    if (radio instanceof Receiver) ((Receiver) radio).accept(new EventHandler(events));
  }

  /**
   * Send a new {@link Request}. Stores transaction-related {@link Request}s if offline. New
   * transaction-related {@link Request}s will be placed behind the queue of stored {@link
   * Request}s.
   *
   * @param uniqueId the id the receiver should use to reply.
   * @param action action name of the {@link Feature}.
   * @param request the outgoing {@link Request}
   */
  public synchronized void sendCall(String uniqueId, String action, Request request) {
    Object call = makeCall(uniqueId, action, packPayload(request));

    try {
      if (radio.isClosed()) {
        if (request.transactionRelated() && transactionQueue != null) {
          logger.warn("Not connected: storing request to queue: {}", request);
          transactionQueue.add(call);
        } else {
          logger.warn("Not connected: can't send request: {}", request);
          events.onError(
              uniqueId,
              "Not connected",
              "The request can't be sent due to the lack of connection",
              request);
        }
      } else if (request.transactionRelated()
          && transactionQueue != null
          && !transactionQueue.isEmpty()) {
        transactionQueue.add(call);
        processTransactionQueue();
      } else {
        radio.send(call);
      }
    } catch (NotConnectedException ex) {
      logger.warn("sendCall() failed: not connected");
      if (request.transactionRelated() && transactionQueue != null) {
        transactionQueue.add(call);
      } else {
        events.onError(
            uniqueId,
            "Not connected",
            "The request can't be sent due to the lack of connection",
            request);
      }
    }
  }

  /**
   * Send a {@link Confirmation} reply to a {@link Request}.
   *
   * @param uniqueId the id the receiver expects.
   * @param action action name of the {@link Feature}.
   * @param confirmation the outgoing {@link Confirmation}
   */
  public void sendCallResult(String uniqueId, String action, Confirmation confirmation) {
    try {
      radio.send(makeCallResult(uniqueId, action, packPayload(confirmation)));

      ConfirmationCompletedHandler completedHandler = confirmation.getCompletedHandler();

      if (completedHandler != null) {
        try {
          completedHandler.onConfirmationCompleted();
        } catch (Throwable e) {
          events.onError(
              uniqueId,
              "ConfirmationCompletedHandlerFailed",
              "The confirmation completed callback handler failed with exception " + e,
              confirmation);
        }
      }
    } catch (NotConnectedException ex) {
      logger.warn("sendCallResult() failed", ex);
      events.onError(
          uniqueId,
          "Not connected",
          "The confirmation couldn't be send due to the lack of connection",
          confirmation);
    }
  }

  /**
   * Send an error. If offline, the message is thrown away.
   *
   * @param uniqueId the id the receiver expects a response to.
   * @param action action name of the {@link Feature}.
   * @param errorCode an OCPP error Code.
   * @param errorDescription a associated error description.
   */
  public void sendCallError(
      String uniqueId, String action, String errorCode, String errorDescription) {
    logger.error(
        "An error occurred. Sending this information: uniqueId {}: action: {}, errorCode: {},"
            + " errorDescription: {}",
        uniqueId,
        action,
        errorCode,
        errorDescription);
    try {
      radio.send(makeCallError(uniqueId, action, errorCode, errorDescription));
    } catch (NotConnectedException ex) {
      logger.warn("sendCallError() failed", ex);
      events.onError(
          uniqueId,
          "Not connected",
          "The error couldn't be send due to the lack of connection",
          errorCode);
    }
  }

  /**
   * Send a call result error. If offline, the message is thrown away.
   *
   * @param uniqueId the id the receiver expects a response to.
   * @param action action name of the {@link Feature}.
   * @param errorCode an OCPP error Code.
   * @param errorDescription a associated error description.
   */
  public void sendCallResultError(
      String uniqueId, String action, String errorCode, String errorDescription) {
    logger.error(
        "An error occurred while processing a call result. Sending this information: "
            + "uniqueId {}: action: {}, errorCode: {}, errorDescription: {}",
        uniqueId,
        action,
        errorCode,
        errorDescription);
    try {
      radio.send(makeCallResultError(uniqueId, action, errorCode, errorDescription));
    } catch (NotConnectedException ex) {
      logger.warn("sendCallResultError() failed", ex);
      events.onError(
          uniqueId,
          "Not connected",
          "The call result error couldn't be sent due to the lack of connection",
          errorCode);
    }
  }

  /**
   * Send a {@link Request} which has no confirmation.
   *
   * @param uniqueId the id of the {@link Request}.
   * @param action action name of the {@link Feature}.
   * @param request the outgoing {@link Request}
   */
  public synchronized void send(String uniqueId, String action, Request request) {
    Object call = makeSend(uniqueId, action, packPayload(request));

    try {
      if (radio.isClosed()) {
        logger.warn("Not connected: can't send request: {}", request);
        events.onError(
            uniqueId,
            "Not connected",
            "The request can't be sent due to the lack of connection",
            request);
      } else {
        radio.send(call);
      }
    } catch (NotConnectedException ex) {
      logger.warn("sendCall() failed: not connected");
      events.onError(
          uniqueId,
          "Not connected",
          "The request can't be sent due to the lack of connection",
          request);
    }
  }

  /** Close down the connection. Uses the {@link Transmitter}. */
  public void disconnect() {
    radio.disconnect();
  }

  private synchronized void processTransactionQueue() {
    if (retryRunner != null && !retryRunner.isAlive()) {
      if (retryRunner.getState() != Thread.State.NEW) {
        retryRunner = new RetryRunner();
      }
      retryRunner.start();
    }
  }

  private class EventHandler implements RadioEvents {
    private final CommunicatorEvents events;

    public EventHandler(CommunicatorEvents events) {
      this.events = events;
    }

    @Override
    public void connected() {
      events.onConnected();
      processTransactionQueue();
    }

    @Override
    public void receivedMessage(Object input) {
      Message message = parse(input);
      if (message instanceof CallResultMessage) {
        events.onCallResult(message.getId(), message.getAction(), message.getPayload());
      } else if (message instanceof CallResultErrorMessage) {
        CallResultErrorMessage call = (CallResultErrorMessage) message;
        events.onCallResultError(
            call.getId(), call.getErrorCode(), call.getErrorDescription(), call.getRawPayload());
      } else if (message instanceof CallErrorMessage) {
        failedFlag.set(true);
        CallErrorMessage call = (CallErrorMessage) message;
        events.onError(
            call.getId(), call.getErrorCode(), call.getErrorDescription(), call.getRawPayload());
      } else if (message instanceof CallMessage) {
        CallMessage call = (CallMessage) message;
        events.onCall(call.getId(), call.getAction(), call.getPayload());
      } else if (message instanceof SendMessage) {
        SendMessage send = (SendMessage) message;
        events.onSend(send.getId(), send.getAction(), send.getPayload());
      }
    }

    @Override
    public void disconnected() {
      events.onDisconnected();
    }
  }

  /**
   * Get queued transaction related request.
   *
   * @return request or null if queue is empty.
   */
  private Object getRetryMessage() {
    Object result = null;
    if (transactionQueue != null && !transactionQueue.isEmpty()) result = transactionQueue.peek();
    return result;
  }

  /**
   * Check if an error message was received.
   *
   * @return whether a fail flag has been raised.
   */
  private boolean hasFailed() {
    return failedFlag.get();
  }

  private void popRetryMessage() {
    if (transactionQueue != null) transactionQueue.pollFirst();
  }

  /** Will resend transaction related requests. */
  private class RetryRunner extends Thread {
    private static final long DELAY_IN_MILLISECONDS = 1000;

    @Override
    public void run() {
      Object call;
      try {
        while ((call = getRetryMessage()) != null) {
          failedFlag.set(false);
          radio.send(call);
          Thread.sleep(DELAY_IN_MILLISECONDS);
          if (!hasFailed()) popRetryMessage();
        }
      } catch (InterruptedException ex) {
        logger.warn("RetryRunner::run() interrupted", ex);
        // restore thread interrupted state
        Thread.currentThread().interrupt();
      } catch (Exception ex) {
        logger.warn("RetryRunner::run() failed", ex);
      }
    }
  }
}