Reset with ongoing transaction (EVerest#128)

* Start with 'reset with ongoing transaction'.

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* Start with database related things to store a transaction message from the queue in the database, so it can be sent after reboot if it is not sended yet.

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* Implement database insert / remove / get functions.

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* Add database base class in common library so the transactions in the queue can be stored in both versions (16 and 201). Put transaction in database as soon as it is received. Remove transaction message from database when reply is received. Some small improvements in for loop with copying / reference.

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* Add evse id to reset callback. Store scheduled 'OnIdle' reset so it an be performed as soon as the charging has stopped.

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* Add evseid to is_reset_allowed_callback

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* Add documentation. Reset when all evse id's stopped charging. 

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* 'initialize' sqlite db member in the constructor. Add documentation

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* Make it possible to call the reset multiple times. Also call the availability handler on reset.

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* Add two templates for conversion from message type to string.

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* Change start() function and add optional boot reason.

Signed-off-by: Maaike <[email protected]>

* Add documentation about start function

Signed-off-by: Maaike <[email protected]>

* sqlite bind text fixes.

Signed-off-by: Maaike <[email protected]>

* Set flag to persist availability after reboot / reset

Signed-off-by: Maaike <[email protected]>

* Add persist to  as well

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* Review comments about race conditions

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* Fill transaction database after bootnotification is receveived and 'accepted'

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* change database handler to shared ptr instead of unique ptr

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>

* clang-format

Signed-off-by: Kai-Uwe Hermann <[email protected]>

---------

Signed-off-by: Maaike Zijderveld, Alfen <[email protected]>
Signed-off-by: Maaike <[email protected]>
Signed-off-by: Kai-Uwe Hermann <[email protected]>
Co-authored-by: Kai-Uwe Hermann <[email protected]>

config/v16/init.sql

@@ -51,3 +51,11 @@ CREATE TABLE IF NOT EXISTS CHARGING_PROFILES (
 CREATE TABLE IF NOT EXISTS OCSP_REQUEST (
     LAST_UPDATE TEXT PRIMARY KEY NOT NULL
 );
+
+CREATE TABLE IF NOT EXISTS TRANSACTION_QUEUE(
+    UNIQUE_ID TEXT PRIMARY KEY NOT NULL,
+    MESSAGE TEXT NOT NULL,
+    MESSAGE_TYPE TEXT NOT NULL,
+    MESSAGE_ATTEMPTS INT NOT NULL,
+    MESSAGE_TIMESTAMP TEXT NOT NULL
+);

config/v201/init_core.sql

@@ -7,4 +7,12 @@ CREATE TABLE IF NOT EXISTS AVAILABILITY(
     EVSE_ID INT PRIMARY KEY NOT NULL,
     CONNECTOR_ID INT,
     OPERATIONAL_STATUS TEXT NOT NULL
-);
+);
+
+CREATE TABLE IF NOT EXISTS TRANSACTION_QUEUE(
+    UNIQUE_ID TEXT PRIMARY KEY NOT NULL,
+    MESSAGE TEXT NOT NULL,
+    MESSAGE_TYPE TEXT NOT NULL,
+    MESSAGE_ATTEMPTS INT NOT NULL,
+    MESSAGE_TIMESTAMP TEXT NOT NULL
+);

include/ocpp/common/database_handler_base.hpp

@@ -0,0 +1,53 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright 2020 - 2023 Pionix GmbH and Contributors to EVerest
+
+#pragma once
+
+#include <deque>
+#include <memory>
+#include <sqlite3.h>
+#include <string>
+
+#include <ocpp/common/types.hpp>
+
+namespace ocpp::common {
+
+struct DBTransactionMessage {
+    json json_message;
+    std::string message_type;
+    int32_t message_attempts;
+    DateTime timestamp;
+    std::string unique_id;
+};
+
+class DatabaseHandlerBase {
+protected:
+    sqlite3* db;
+
+public:
+    ///
+    /// \brief Base database handler class.
+    ///
+    /// Other database classes should derive from this class and only use the functions after `db` is initialized.
+    /// Class handles some common database functionality like inserting and removing transaction messages.
+    ///
+    /// \warning The 'db' variable is not initialized, the deriving class should do that.
+    ///
+    DatabaseHandlerBase() noexcept;
+
+    /// \brief Get transaction messages from transaction messages queue table.
+    /// \return The transaction messages.
+    std::vector<DBTransactionMessage> get_transaction_messages();
+
+    /// \brief Insert a new transaction message that needs to be sent to the CSMS.
+    /// \param transaction_message  The message to be stored.
+    /// \return True on success.
+    bool insert_transaction_message(const DBTransactionMessage& transaction_message);
+
+    /// \brief Remove a transaction message from the database.
+    /// \param unique_id    The unique id of the transaction message.
+    /// \return True on success.
+    void remove_transaction_message(const std::string& unique_id);
+};
+
+} // namespace ocpp::common

include/ocpp/common/message_queue.hpp

@@ -18,6 +18,7 @@
 #include <everest/timer.hpp>
 
 #include <ocpp/common/call_types.hpp>
+#include <ocpp/common/database_handler_base.hpp>
 #include <ocpp/common/types.hpp>
 #include <ocpp/v16/types.hpp>
 #include <ocpp/v201/types.hpp>
@@ -56,22 +57,23 @@ template <typename M> struct ControlMessage {
 
     /// \brief Provides the unique message ID stored in the message
     /// \returns the unique ID of the contained message
-    MessageId uniqueId() {
+    MessageId uniqueId() const {
         return this->message[MESSAGE_ID];
     }
 };
 
 /// \brief contains a message queue that makes sure that OCPPs synchronicity requirements are met
 template <typename M> class MessageQueue {
 private:
+    std::shared_ptr<ocpp::common::DatabaseHandlerBase> database_handler;
     int transaction_message_attempts;
     int transaction_message_retry_interval; // seconds
     std::thread worker_thread;
     /// message deque for transaction related messages
-    std::deque<ControlMessage<M>*> transaction_message_queue;
+    std::deque<std::shared_ptr<ControlMessage<M>>> transaction_message_queue;
     /// message queue for non-transaction related messages
-    std::queue<ControlMessage<M>*> normal_message_queue;
-    ControlMessage<M>* in_flight;
+    std::queue<std::shared_ptr<ControlMessage<M>>> normal_message_queue;
+    std::shared_ptr<ControlMessage<M>> in_flight;
     std::mutex message_mutex;
     std::condition_variable cv;
     std::function<bool(json message)> send_callback;
@@ -121,9 +123,9 @@ template <typename M> class MessageQueue {
         return false;
     }
 
-    bool isTransactionMessage(ControlMessage<M>* message);
+    bool isTransactionMessage(const std::shared_ptr<ControlMessage<M>> message) const;
 
-    void add_to_normal_message_queue(ControlMessage<M>* message) {
+    void add_to_normal_message_queue(std::shared_ptr<ControlMessage<M>> message) {
         EVLOG_debug << "Adding message to normal message queue";
         {
             std::lock_guard<std::mutex> lk(this->message_mutex);
@@ -133,11 +135,15 @@ template <typename M> class MessageQueue {
         this->cv.notify_all();
         EVLOG_debug << "Notified message queue worker";
     }
-    void add_to_transaction_message_queue(ControlMessage<M>* message) {
+    void add_to_transaction_message_queue(std::shared_ptr<ControlMessage<M>> message) {
         EVLOG_debug << "Adding message to transaction message queue";
         {
             std::lock_guard<std::mutex> lk(this->message_mutex);
             this->transaction_message_queue.push_back(message);
+            ocpp::common::DBTransactionMessage db_message{message->message, messagetype_to_string(message->messageType),
+                                                          message->message_attempts, message->timestamp,
+                                                          message->uniqueId()};
+            this->database_handler->insert_transaction_message(db_message);
             this->new_message = true;
         }
         this->cv.notify_all();
@@ -147,14 +153,17 @@ template <typename M> class MessageQueue {
 public:
     /// \brief Creates a new MessageQueue object with the provided \p configuration and \p send_callback
     MessageQueue(const std::function<bool(json message)>& send_callback, const int transaction_message_attempts,
-                 const int transaction_message_retry_interval, const std::vector<M>& external_notify) :
+                 const int transaction_message_retry_interval, const std::vector<M>& external_notify,
+                 std::shared_ptr<common::DatabaseHandlerBase> database_handler) :
+        database_handler(database_handler),
         transaction_message_attempts(transaction_message_attempts),
         transaction_message_retry_interval(transaction_message_retry_interval),
         external_notify(external_notify),
         paused(true),
         running(true),
         new_message(false),
         uuid_generator(boost::uuids::random_generator()) {
+
         this->send_callback = send_callback;
         this->in_flight = nullptr;
         this->worker_thread = std::thread([this]() {
@@ -190,7 +199,7 @@ template <typename M> class MessageQueue {
 
                 // prioritize the message with the oldest timestamp
                 auto now = DateTime();
-                ControlMessage<M>* message = nullptr;
+                std::shared_ptr<ControlMessage<M>> message = nullptr;
                 QueueType queue_type = QueueType::None;
 
                 if (!this->normal_message_queue.empty()) {
@@ -289,12 +298,33 @@ template <typename M> class MessageQueue {
     }
 
     MessageQueue(const std::function<bool(json message)>& send_callback, const int transaction_message_attempts,
-                 const int transaction_message_retry_interval) :
-        MessageQueue(send_callback, transaction_message_attempts, transaction_message_retry_interval, {}){};
+                 const int transaction_message_retry_interval,
+                 std::shared_ptr<common::DatabaseHandlerBase> databaseHandler) :
+        MessageQueue(send_callback, transaction_message_attempts, transaction_message_retry_interval, {},
+                     databaseHandler) {
+    }
+
+    void get_transaction_messages_from_db() {
+        std::vector<ocpp::common::DBTransactionMessage> transaction_messages =
+            database_handler->get_transaction_messages();
+
+        if (!transaction_messages.empty()) {
+            for (auto& transaction_message : transaction_messages) {
+                std::shared_ptr<ControlMessage<M>> message =
+                    std::make_shared<ControlMessage<M>>(transaction_message.json_message);
+                message->messageType = string_to_messagetype(transaction_message.message_type);
+                message->timestamp = transaction_message.timestamp;
+                message->message_attempts = transaction_message.message_attempts;
+                transaction_message_queue.push_back(message);
+            }
+
+            this->new_message = true;
+        }
+    }
 
     /// \brief pushes a new \p call message onto the message queue
     template <class T> void push(Call<T> call) {
-        auto* message = new ControlMessage<M>(call);
+        auto message = std::make_shared<ControlMessage<M>>(call);
         if (this->isTransactionMessage(message)) {
             // according to the spec the "transaction related messages" StartTransaction, StopTransaction and
             // MeterValues have to be delivered in chronological order
@@ -315,7 +345,7 @@ template <typename M> class MessageQueue {
     /// \brief pushes a new \p call message onto the message queue
     /// \returns a future from which the CallResult can be extracted
     template <class T> std::future<EnhancedMessage<M>> push_async(Call<T> call) {
-        auto* message = new ControlMessage<M>(call);
+        auto message = std::make_shared<ControlMessage<M>>(call);
         if (this->isTransactionMessage(message)) {
             // according to the spec the "transaction related messages" StartTransaction, StopTransaction and
             // MeterValues have to be delivered in chronological order
@@ -398,6 +428,13 @@ template <typename M> class MessageQueue {
             enhanced_message.messageType = this->string_to_messagetype(
                 this->in_flight->message.at(CALL_ACTION).template get<std::string>() + std::string("Response"));
             this->in_flight->promise.set_value(enhanced_message);
+
+            if (isTransactionMessage(this->in_flight)) {
+                // We only remove the message as soon as a response is received. Otherwise we might miss a message if
+                // the charging station just boots after sending, but before receiving the result.
+                this->database_handler->remove_transaction_message(this->in_flight->uniqueId());
+            }
+
             this->reset_in_flight();
 
             // we want the start transaction response handler to be executed before the next message will be
@@ -557,6 +594,7 @@ template <typename M> class MessageQueue {
     }
 
     M string_to_messagetype(const std::string& s);
+    std::string messagetype_to_string(M m);
 };
 
 } // namespace ocpp

include/ocpp/v16/database_handler.hpp

@@ -8,6 +8,7 @@
 #include <fstream>
 #include <iostream>
 
+#include <ocpp/common/database_handler_base.hpp>
 #include <ocpp/common/schemas.hpp>
 #include <ocpp/common/types.hpp>
 #include <ocpp/v16/ocpp_types.hpp>
@@ -37,9 +38,8 @@ struct TransactionEntry {
 };
 
 /// \brief This class handles the connection and operations of the SQLite database
-class DatabaseHandler {
+class DatabaseHandler : public ocpp::common::DatabaseHandlerBase {
 private:
-    sqlite3* db;
     std::filesystem::path db_path;          // directory where the database file is located
     std::filesystem::path init_script_path; // full path of init sql script
 

include/ocpp/v201/charge_point.hpp

@@ -2,6 +2,7 @@
 // Copyright 2020 - 2023 Pionix GmbH and Contributors to EVerest
 
 #include <future>
+#include <set>
 
 #include <ocpp/common/charging_station_base.hpp>
 
@@ -42,11 +43,25 @@ namespace ocpp {
 namespace v201 {
 
 struct Callbacks {
-    std::function<bool(const ResetEnum& reset_type)> is_reset_allowed_callback;
-    std::function<void(const ResetEnum& reset_type)> reset_callback;
+    ///
+    /// \brief Callback if reset is allowed. If evse_id has a value, reset only applies to the given evse id. If it has
+    ///        no value, applies to complete charging station.
+    ///
+    std::function<bool(const std::optional<const int32_t> evse_id, const ResetEnum& reset_type)>
+        is_reset_allowed_callback;
+    std::function<void(const std::optional<const int32_t> evse_id, const ResetEnum& reset_type)> reset_callback;
     std::function<void(const int32_t evse_id, const ReasonEnum& stop_reason)> stop_transaction_callback;
     std::function<void(const int32_t evse_id)> pause_charging_callback;
-    std::function<void(const ChangeAvailabilityRequest& request)> change_availability_callback;
+    ///
+    /// \brief Change availability of charging station / evse / connector.
+    /// \param request The request.
+    /// \param persist True to persist the status after reboot.
+    ///
+    /// Persist is set to 'false' if the status does not need to be stored after restarting. Otherwise it is true.
+    /// False is for example during a reset OnIdle where first an 'unavailable' is sent until the charging session
+    /// stopped. True is for example when the CSMS sent an 'inoperative' request.
+    ///
+    std::function<void(const ChangeAvailabilityRequest& request, const bool persist)> change_availability_callback;
     std::function<GetLogResponse(const GetLogRequest& request)> get_log_request_callback;
     std::function<UnlockConnectorResponse(const int32_t evse_id, const int32_t connecor_id)> unlock_connector_callback;
     // callback to be called when the request can be accepted. authorize_remote_start indicates if Authorize.req needs
@@ -82,7 +97,7 @@ class ChargePoint : ocpp::ChargingStationBase {
     // utility
     std::unique_ptr<MessageQueue<v201::MessageType>> message_queue;
     std::unique_ptr<DeviceModel> device_model;
-    std::unique_ptr<DatabaseHandler> database_handler;
+    std::shared_ptr<DatabaseHandler> database_handler;
 
     std::map<int32_t, ChangeAvailabilityRequest> scheduled_change_availability_requests;
 
@@ -106,6 +121,11 @@ class ChargePoint : ocpp::ChargingStationBase {
     int network_configuration_priority;
     bool disable_automatic_websocket_reconnects;
 
+    /// \brief Used when an 'OnIdle' reset is requested, to perform the reset after the charging has stopped.
+    bool reset_scheduled;
+    /// \brief If `reset_scheduled` is true and the reset is for a specific evse id, it will be stored in this member.
+    std::set<int32_t> reset_scheduled_evseids;
+
     // callback struct
     Callbacks callbacks;
 
@@ -165,6 +185,14 @@ class ChargePoint : ocpp::ChargingStationBase {
     ///
     bool is_evse_connector_available(const std::unique_ptr<Evse>& evse) const;
 
+    ///
+    /// \brief Set all connectors of a given evse to unavailable.
+    /// \param evse The evse.
+    /// \param persist  True if unavailability should persist. If it is set to false, there will be a check per
+    ///                 connector if it was already set to true and if that is the case, it will be persisted anyway.
+    ///
+    void set_evse_connectors_unavailable(const std::unique_ptr<Evse>& evse, bool persist);
+
     /* OCPP message requests */
 
     // Functional Block B: Provisioning
@@ -245,7 +273,8 @@ class ChargePoint : ocpp::ChargingStationBase {
                 const std::string& message_log_path, const std::string& certs_path, const Callbacks& callbacks);
 
     /// \brief Starts the ChargePoint, initializes and connects to the Websocket endpoint
-    void start();
+    /// \param bootreason   Optional bootreason (default: PowerUp).
+    void start(BootReasonEnum bootreason = BootReasonEnum::PowerUp);
 
     /// \brief Starts the websocket
     void start_websocket();
@@ -264,7 +293,7 @@ class ChargePoint : ocpp::ChargingStationBase {
     /// \param request_id   The request_id. When it is -1, it will not be included in the request.
     /// \param firmware_update_status The firmware_update_status should be convertable to the
     ///        ocpp::v201::FirmwareStatusEnum.
-    void on_firmware_update_status_notification(int32_t request_id, std::string& firmware_update_status);
+    void on_firmware_update_status_notification(int32_t request_id, const std::string& firmware_update_status);
 
     /// \brief Event handler that should be called when a session has started
     /// \param evse_id

include/ocpp/v201/database_handler.hpp

@@ -4,10 +4,12 @@
 #define OCPP_V201_DATABASE_HANDLER_HPP
 
 #include "sqlite3.h"
+#include <deque>
 #include <filesystem>
 #include <fstream>
 #include <memory>
 
+#include <ocpp/common/database_handler_base.hpp>
 #include <ocpp/v201/ocpp_types.hpp>
 
 #include <everest/logging.hpp>
@@ -17,10 +19,9 @@ namespace fs = std::filesystem;
 namespace ocpp {
 namespace v201 {
 
-class DatabaseHandler {
+class DatabaseHandler : public ocpp::common::DatabaseHandlerBase {
 
 private:
-    sqlite3* db;
     fs::path database_file_path;
     fs::path sql_init_path;
 

include/ocpp/v201/evse.hpp

@@ -47,7 +47,7 @@ class Evse {
 
     /// \brief Returns the number of connectors of this EVSE
     /// \return
-    int32_t get_number_of_connectors();
+    uint32_t get_number_of_connectors();
 
     /// \brief Returns a reference to the sampled meter values timer
     /// \return

lib/CMakeLists.txt

@@ -5,6 +5,7 @@ target_sources(ocpp
     PRIVATE
         ocpp/common/call_types.cpp
         ocpp/common/charging_station_base.cpp
+        ocpp/common/database_handler_base.cpp
         ocpp/common/message_queue.cpp
         ocpp/common/ocpp_logging.cpp
         ocpp/common/pki_handler.cpp