publisher.h

Go to the documentation of this file.

/*
 * Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License").
 * You may not use this file except in compliance with the License.
 * A copy of the License is located at
 *
 *  http://aws.amazon.com/apache2.0
 *
 * or in the "license" file accompanying this file. This file 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.
 */

#pragma once

#include <atomic>
#include <chrono>
#include <list>
#include <functional>
#include <iostream>
#include <mutex>

#include <dataflow_lite/utils/observable_object.h>
#include <dataflow_lite/utils/service.h>
#include <dataflow_lite/task/task.h>

enum PublisherState {
    UNKNOWN, // constructed
    CONNECTED, //configured, ready to receive data
    NOT_CONNECTED, //unable to send data
};

template <typename T>
class Publisher : public Aws::DataFlow::IPublisher<T>, public Service
{

public:

    Publisher() : publisher_state_(UNKNOWN) {
      publish_successes_.store(0);
      publish_attempts_.store(0);
      last_publish_duration_.store(std::chrono::milliseconds(0));
    }

    ~Publisher() override = default;

    PublisherState getPublisherState() {
      return publisher_state_.getValue();
    }

    Aws::DataFlow::UploadStatus attemptPublish(T &data) override {

      // don't attempt to publish if not in the started state
      if(ServiceState::STARTED != this->getState()) {
        return Aws::DataFlow::UploadStatus::FAIL;
      }

      publish_attempts_++;
      auto published_status = Aws::DataFlow::UploadStatus::FAIL;

      // acquire lock to publish
      std::lock_guard<std::mutex> lck (publish_mutex_);
      {
        auto start = std::chrono::high_resolution_clock::now();
        published_status = publishData(data); // always at least try
        last_publish_duration_.store(std::chrono::duration_cast<std::chrono::milliseconds>(
                std::chrono::high_resolution_clock::now() - start));
      }

      if (Aws::DataFlow::UploadStatus::SUCCESS == published_status) {
        publish_successes_++;
        publisher_state_.setValue(CONNECTED);
      } else {
        publisher_state_.setValue(NOT_CONNECTED);
      }

      return published_status;
    }

    bool shutdown() override {
      bool b = Service::shutdown();  // set shutdown state to try and fast fail any publish calls

      std::lock_guard<std::mutex> lck (publish_mutex_);
      //acquire the lock to ensure attemptPublish has finished
      publisher_state_.setValue(UNKNOWN);
      return b;
    }

    bool canPublish() {
      auto current_state = publisher_state_.getValue();
      return (current_state == UNKNOWN || current_state == CONNECTED) && ServiceState::STARTED == this->getState();
    }

    int getPublishSuccesses() {
      return publish_successes_.load();
    }

    int getPublishAttempts() {
      return publish_attempts_.load();
    }

    std::chrono::milliseconds getLastPublishDuration() {
      return last_publish_duration_.load();
    }

    float getPublishSuccessPercentage() {
      int attempts = publish_attempts_.load();
      if (attempts == 0) {
        return 0;
      }
      int successes = publish_successes_.load();
      return static_cast<float>(successes) / static_cast<float>(attempts) * 100.0f;
    }

    virtual void addPublisherStateListener(const std::function<void(const PublisherState&)> & listener) {
      publisher_state_.addListener(listener);
    }

protected:

    virtual Aws::DataFlow::UploadStatus publishData(T &data) = 0;

private:
    ObservableObject<PublisherState> publisher_state_;
    std::atomic<int> publish_successes_{};
    std::atomic<int> publish_attempts_{};
    std::atomic<std::chrono::milliseconds> last_publish_duration_{};
    mutable std::mutex publish_mutex_;

};