text: /*
   * #%L
   * %%
   * Copyright (C) 2020 BMW Car IT GmbH
   * %%
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   * 
   *      http://www.apache.org/licenses/LICENSE-2.0
   * 
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   * #L%
   */
  #include "joynr/UdsClient.h"
  
  #include <string>
  
  #include "joynr/Util.h"
  
  #include "UdsFrameBufferV1.h"
  #include "UdsSendQueue.h"
  
  namespace joynr
  {
  
  constexpr int UdsClient::_threadsPerConnection;
  
  UdsClient::UdsClient(
          const UdsSettings& settings,
          const std::function<void(const exceptions::JoynrRuntimeException&)>& onFatalRuntimeError)
          : IUdsSender(),
            _fatalRuntimeErrorCallback{onFatalRuntimeError},
            _connectedCallback{[]() {}},
            _disconnectedCallback{[]() {}},
            _receivedCallback{[](smrf::ByteVector&&) {}},
            _address{settings.createClientMessagingAddress()},
            _connectSleepTime{settings.getConnectSleepTimeMs()},
            _sendQueue(
                    std::make_unique<UdsSendQueue<UdsFrameBufferV1>>(settings.getSendingQueueSize())),
            _readBuffer(std::make_unique<UdsFrameBufferV1>()),
            _endpoint(settings.getSocketPath()),
            _ioContext(_threadsPerConnection),
            _socket(_ioContext),
            _state{State::STOP}
  {
      try {
          _sendQueue->pushBack(UdsFrameBufferV1(_address));
      } catch (const std::exception& e) {
          doHandleFatalError("Failed to insert INIT message to queue.", e);
      }
  }
  
  UdsClient::~UdsClient()
  {
      shutdown();
  }
  
  joynr::system::RoutingTypes::UdsClientAddress UdsClient::getAddress() const noexcept
  {
      return _address;
  }
  
  void UdsClient::setConnectCallback(const Connected& callback)
  {
      if (callback) {
          _connectedCallback = callback;
      }
  }
  
  void UdsClient::setDisconnectCallback(const Disconnected& callback)
  {
      if (callback) {
          _disconnectedCallback = callback;
      }
  }
  
  void UdsClient::setReceiveCallback(const Received& callback)
  {
      if (callback) {
          _receivedCallback = callback;
      }
  }
  
  void UdsClient::start()
  {
      if (_worker.valid()) {
          JOYNR_LOG_ERROR(logger(), "{} alreay started.", _address.getId());
          return;
      }
      _state.store(State::START);
      _worker = std::async(&UdsClient::run, this);
  }
  
  void UdsClient::shutdown() noexcept
  {
      std::lock_guard<std::mutex> lockStop(_asyncShutdownMutex);
      _state.store(State::STOP);
      try {
          _ioContext.stop();
          if (_worker.valid()) {
              _worker.get();
          }
      } catch (std::exception& e) {
          JOYNR_LOG_ERROR(logger(),
                          "{} caused unexpected error when stopping worker: {}",
                          _address.getId(),
                          e.what());
      } catch (...) {
          JOYNR_LOG_ERROR(logger(), "{} caused unknown error when stopping worker", _address.getId());
      }
  }
  
  void UdsClient::run()
  {
      bool isRetry = false;
      while (State::START == _state.load()) {
          if (isRetry) {
              // Wait before retry connection so that worker does not block everything
              std::this_thread::sleep_for(_connectSleepTime);
              _ioContext.reset();
          }
          isRetry = true;
          _socket.async_connect(_endpoint, [this](boost::system::error_code failedToConnect) {
              if (failedToConnect) {
                  JOYNR_LOG_ERROR(logger(),
                                  "{} ({}) failed to connect. Retry in {:d}ms.",
                                  _address.getId(),
                                  _endpoint.path(),
                                  _connectSleepTime.count());
              } else {
                  if (State::START == _state.exchange(State::CONNECTED)) {
                      try {
                          _connectedCallback();
                      } catch (const std::exception& e) {
                          doHandleFatalError("Error from connect callback", e);
                      }
                      _ioContext.post([this]() {
                          // Send messages currently in the queue, which contains at least the initial
                          // message
                          doWrite();
                      });
                      doReadHeader();
                  }
              }
          });
          _ioContext.run();
          _socket.close();
      }
      if (State::CONNECTED == _state.load()) {
          doHandleFatalError("State machine stopped unexpectedly");
      }
      try {
          _disconnectedCallback();
      } catch (const std::exception& e) {
          doHandleFatalError("Failed to process disconnection", e);
      }
  }
  
  void UdsClient::doReadHeader() noexcept
  {
      boost::asio::async_read(_socket,
                              _readBuffer->header(),
                              [this](boost::system::error_code readFailure, std::size_t /*length*/) {
          if (readFailure) {
              JOYNR_LOG_ERROR(logger(),
                              "{} failed to read header: {}",
                              _address.getId(),
                              readFailure.message());
          } else {
              doReadBody();
          }
      });
  }
  
  void UdsClient::doReadBody() noexcept
  {
      try {
          boost::asio::async_read(
                  _socket,
                  _readBuffer->body(),
                  [this](boost::system::error_code readFailure, std::size_t /*length*/) {
                      if (readFailure) {
                          JOYNR_LOG_ERROR(logger(),
                                          "{} failed to read body: {}",
                                          _address.getId(),
                                          readFailure.message());
                      } else {
                          try {
                              _receivedCallback(_readBuffer->readMessage());
                              doReadHeader();
                          } catch (const std::exception& e) {
                              doHandleFatalError("Failed to process message-frame", e);
                          }
                      }
                  });
      } catch (const std::exception& e) {
          doHandleFatalError("Failed to read message-frame", e);
      }
  }
  
  void UdsClient::send(const smrf::ByteArrayView& msg, const IUdsSender::SendFailed& callback)
  {
      try {
          _ioContext.post([ this, frame = UdsFrameBufferV1(msg), callback ]() mutable {
              try {
                  if (_sendQueue->pushBack(std::move(frame), callback)) {
                      doWrite();
                  }
              } catch (const std::exception& e) {
                  doHandleFatalError("Failed to queue message", e);
              }
          });
      } catch (const std::exception& e) {
          doHandleFatalError("Failed to create message frame", e);
      }
  }
  
  void UdsClient::doWrite() noexcept
  {
      boost::asio::async_write(_socket,
                               _sendQueue->showFront(),
                               [this](boost::system::error_code writeFailed, std::size_t /*length*/) {
          if (_sendQueue->popFrontOnSuccess(writeFailed)) {
              doWrite();
          }
      });
  }
  
  void UdsClient::doHandleFatalError(const std::string& errorMessage,
                                     const std::exception& error) noexcept
  {
      doHandleFatalError(errorMessage + " - " + error.what());
  }
  
  void UdsClient::doHandleFatalError(const std::string& errorMessage) noexcept
  {
      JOYNR_LOG_FATAL(logger(),
                      "{} fatal runtime error, stopping all communication permanently: {}",
                      _address.getId(),
                      errorMessage);
      if (!hasFatalErrorAlreadyBeenReported()) {
          try {
              if (_fatalRuntimeErrorCallback) {
                  _fatalRuntimeErrorCallback(exceptions::JoynrRuntimeException(errorMessage));
              }
          } catch (std::exception& e) {
              JOYNR_LOG_ERROR(logger(),
                              "{} received unexpected exception when informing user about "
                              "internal fatal runtime error: {}",
                              _address.getId(),
                              e.what());
          } catch (...) {
              JOYNR_LOG_ERROR(logger(),
                              "{} received an unkown error when informing user about "
                              "internal fatal runtime error.",
                              _address.getId());
          }
          boost::system::error_code ignore; // Maybe the socket is not open.
          _socket.close(ignore);
      }
  }
  
  bool UdsClient::hasFatalErrorAlreadyBeenReported() noexcept
  {
      const auto previous = _state.exchange(State::FAILED);
      switch (previous) {
      case State::FAILED:
          // This call is a follow-up of previously reported error
          return true;
      case State::STOP:
          // In case of shutdown, don't report a connection problem as error
          return true;
      case State::CONNECTED:
      case State::START:
          break;
      }
      return false;
  }
  
  } // namespace joynr
  