kinesis_stream_manager.h

Go to the documentation of this file.

/*
 * Copyright 2018 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 <aws/kinesis/KinesisClient.h>
#include <aws/kinesis/model/GetRecordsRequest.h>
#include <aws/kinesis/model/GetRecordsResult.h>
#include <aws/kinesis/model/Record.h>
#include <aws_common/sdk_utils/client_configuration_provider.h>
#include <aws_common/sdk_utils/logging/aws_log_system.h>
#include <kinesis-video-producer/KinesisVideoProducer.h>
#include <kinesis_manager/common.h>
#include <kinesis_manager/kinesis_client_facade.h>
#include <kinesis_manager/stream_definition_provider.h>
#include <kinesis_manager/stream_subscription_installer.h>
#include <kinesis_manager/kinesis_video_producer_interface.h>

namespace Aws {
namespace Kinesis {

inline Aws::Client::ParameterPath GetStreamParameterPath(int stream_idx, const char * parameter_name)
{
  Aws::Client::ParameterPath path(kStreamParameters.prefix);
  if (INVALID_STREAM_ID != stream_idx) {
    path += kStreamParameters.stream_namespace + std::to_string(stream_idx);
  }
  if (nullptr != parameter_name) {
    path += parameter_name;
  }
  return path;
}

inline Aws::Client::ParameterPath GetStreamParameterPrefix(int stream_idx)
{
  return GetStreamParameterPath(stream_idx, nullptr);
}
inline Aws::Client::ParameterPath GetKinesisVideoParameter(const char * parameter_name)
{
  return GetStreamParameterPath(INVALID_STREAM_ID, parameter_name);
}

class KinesisStreamManagerInterface
{
public:
  KinesisStreamManagerInterface(const Aws::Client::ParameterReaderInterface * parameter_reader,
                                const StreamDefinitionProvider * stream_definition_provider,
                                StreamSubscriptionInstaller * subscription_installer)
  : parameter_reader_(parameter_reader),
    stream_definition_provider_(stream_definition_provider),
    subscription_installer_(subscription_installer){};
  KinesisStreamManagerInterface() = default;
  virtual ~KinesisStreamManagerInterface() = default;

  using VideoProducerFactory = std::function<std::unique_ptr<KinesisVideoProducerInterface>(
    std::string,
    std::unique_ptr<com::amazonaws::kinesis::video::DeviceInfoProvider>,
    std::unique_ptr<com::amazonaws::kinesis::video::ClientCallbackProvider>,
    std::unique_ptr<com::amazonaws::kinesis::video::StreamCallbackProvider>,
    std::unique_ptr<com::amazonaws::kinesis::video::CredentialProvider>
  )>;

  virtual KinesisManagerStatus InitializeVideoProducer(std::string region,
    std::unique_ptr<com::amazonaws::kinesis::video::DeviceInfoProvider> device_info_provider,
    std::unique_ptr<com::amazonaws::kinesis::video::ClientCallbackProvider> client_callback_provider,
    std::unique_ptr<com::amazonaws::kinesis::video::StreamCallbackProvider> stream_callback_provider,
    std::unique_ptr<com::amazonaws::kinesis::video::CredentialProvider> credential_provider,
    VideoProducerFactory video_producer_factory = KinesisStreamManagerInterface::CreateDefaultVideoProducer) = 0;

  virtual KinesisManagerStatus InitializeVideoProducer(std::string region,
    VideoProducerFactory video_producer_factory = KinesisStreamManagerInterface::CreateDefaultVideoProducer) = 0;

  virtual KinesisManagerStatus InitializeVideoStream(
    std::unique_ptr<com::amazonaws::kinesis::video::StreamDefinition> stream_definition) = 0;

  virtual KinesisManagerStatus PutFrame(std::string stream_name, Frame & frame) const = 0;

  virtual KinesisManagerStatus PutMetadata(std::string stream_name, const std::string & name,
                                           const std::string & value) const = 0;

  virtual void FreeStream(std::string stream_name) = 0;

  virtual KinesisManagerStatus ProcessCodecPrivateDataForStream(
    const std::string & stream_name, std::vector<uint8_t> codec_private_data) = 0;

  virtual KinesisManagerStatus FetchRekognitionResults(const std::string & stream_name,
                                                       Aws::Vector<Model::Record> * records) = 0;

  static std::unique_ptr<KinesisVideoProducerInterface> CreateDefaultVideoProducer(
    std::string region,
    std::unique_ptr<com::amazonaws::kinesis::video::DeviceInfoProvider> device_info_provider,
    std::unique_ptr<com::amazonaws::kinesis::video::ClientCallbackProvider> client_callback_provider,
    std::unique_ptr<com::amazonaws::kinesis::video::StreamCallbackProvider> stream_callback_provider,
    std::unique_ptr<com::amazonaws::kinesis::video::CredentialProvider> credential_provider);

protected:
  virtual KinesisManagerStatus KinesisVideoStreamerSetup();
  virtual KinesisManagerStatus KinesisVideoStreamSetup(const uint16_t stream_idx,
                                                       const PBYTE codec_private_data,
                                                       const uint32_t codec_private_data_size,
                                                       std::string * stream_name);
  virtual KinesisManagerStatus GenerateStreamSubscriptionDescriptor(
    int stream_idx, StreamSubscriptionDescriptor & descriptor);
  virtual KinesisManagerStatus InitializeStreamSubscription(
    const StreamSubscriptionDescriptor & descriptor) = 0;

  const Aws::Client::ParameterReaderInterface * parameter_reader_ = nullptr;
  const StreamDefinitionProvider * stream_definition_provider_ = nullptr;
  StreamSubscriptionInstaller * subscription_installer_ = nullptr;
};

class KinesisStreamManager : public KinesisStreamManagerInterface
{
public:
  KinesisStreamManager(Aws::Client::ParameterReaderInterface * parameter_reader,
                       StreamDefinitionProvider * stream_definition_provider,
                       StreamSubscriptionInstaller * subscription_installer,
                       std::unique_ptr<KinesisClient> kinesis_client)
  : kinesis_client_(std::move(kinesis_client)),
    KinesisStreamManagerInterface(parameter_reader, stream_definition_provider,
                                  subscription_installer){};
  KinesisStreamManager() = default;
  ~KinesisStreamManager() = default;

  KinesisManagerStatus InitializeVideoProducer(std::string region,
    std::unique_ptr<com::amazonaws::kinesis::video::DeviceInfoProvider> device_info_provider,
    std::unique_ptr<com::amazonaws::kinesis::video::ClientCallbackProvider> client_callback_provider,
    std::unique_ptr<com::amazonaws::kinesis::video::StreamCallbackProvider> stream_callback_provider,
    std::unique_ptr<com::amazonaws::kinesis::video::CredentialProvider> credential_provider,
    KinesisStreamManagerInterface::VideoProducerFactory video_producer_factory = KinesisStreamManagerInterface::CreateDefaultVideoProducer) override;
  KinesisManagerStatus InitializeVideoProducer(std::string region,
    KinesisStreamManagerInterface::VideoProducerFactory video_producer_factory = KinesisStreamManagerInterface::CreateDefaultVideoProducer) override;

  KinesisManagerStatus InitializeVideoStream(
    std::unique_ptr<com::amazonaws::kinesis::video::StreamDefinition> stream_definition) override;

  KinesisManagerStatus PutFrame(std::string stream_name, Frame & frame) const override;

  KinesisManagerStatus PutMetadata(std::string stream_name, const std::string & name,
                                   const std::string & value) const override;

  void FreeStream(std::string stream_name) override;

  KinesisManagerStatus KinesisVideoStreamerSetup() override
  {
    return KinesisStreamManagerInterface::KinesisVideoStreamerSetup();
  }

  KinesisManagerStatus KinesisVideoStreamSetup(const uint16_t stream_idx,
                                               const PBYTE codec_private_data,
                                               const uint32_t codec_private_data_size,
                                               std::string * stream_name) override
  {
    return KinesisStreamManagerInterface::KinesisVideoStreamSetup(
      stream_idx, codec_private_data, codec_private_data_size, stream_name);
  }

  KinesisManagerStatus ProcessCodecPrivateDataForStream(
    const std::string & stream_name, std::vector<uint8_t> codec_private_data) override;

  KinesisManagerStatus FetchRekognitionResults(const std::string & stream_name,
                                               Aws::Vector<Model::Record> * records) override;

  KinesisVideoProducerInterface * get_video_producer()
  {
    return video_producer_.get();
  }

protected:
  KinesisManagerStatus InitializeStreamSubscription(
    const StreamSubscriptionDescriptor & descriptor) override;

private:
  KinesisManagerStatus UpdateShardIterator(const std::string & stream_name);

  std::map<std::string, std::shared_ptr<KinesisVideoStreamInterface>> video_streams_;
  std::map<std::string, std::vector<uint8_t>> video_streams_codec_data_;
  std::unique_ptr<KinesisVideoProducerInterface> video_producer_;
  std::unique_ptr<KinesisClient> kinesis_client_;

  struct RekognitionStreamInfo
  {
    Aws::String data_stream_name; /* Analysis results will be fetched from this stream. */
    Aws::String shard_iterator;
  };
  std::map<std::string, RekognitionStreamInfo>
    rekognition_config_; /* Video stream name to RekognitionStreamInfo */
};

}  // namespace Kinesis
}  // namespace Aws