ReducingHandler.hpp

Go to the documentation of this file.

#ifndef REDUCINGHANDLER_HPP_
#define REDUCINGHANDLER_HPP_

#include "ReducingPartsCountersStorageIf.hpp"
#include "ReduceAdditionsStorageIf.hpp"
#include "TaskReducersManagerFactoryIf.hpp"
#include "ReducingOutputMessageBuilderIf.hpp"
#include "ReducingInputMessage.hpp"
#include "SphinxReduceJobBuilder.hpp"
#include "SupportNotifierIf.hpp"
#include "ReducingTaskTTLManager.hpp"
#include <Poco/SharedPtr.h>

namespace HCE{
namespace reduce{

class ReducingHandler{
public:
  ReducingHandler(
    const Poco::SharedPtr<ReducingPartsCountersStorageIf>& reducingPartsCountersStorage,
    const Poco::SharedPtr<ReduceAdditionsStorageIf>& reduceAdditionsStorage,
    const Poco::SharedPtr<TaskReducersManagerFactoryIf> &taskReducersManagerFactory,
    const Poco::SharedPtr<ReducingOutputMessageBuilderIf>& reducingOutputMessageBuilder,
    const Poco::SharedPtr<SupportNotifierIf>& supportNotifier);
  virtual ~ReducingHandler();

  void accumulateReducingData(unsigned long long itemId,
       Poco::SharedPtr<reduce_types::ReducingInputMessage>& reducingInputMessage);
  Poco::SharedPtr<reduce_types::ReducingOutputMessage> makeReducing(unsigned long long itemId);
  void deleteReducingTaskBy(unsigned long long itemId);

  int getAccumulateReducingPartsNumberBy(unsigned long long itemId);
  unsigned int getTasksNumber();
  bool isTaskTerminatedByTTL( unsigned long long taskId ) const;
  bool isTaskExpired( unsigned long long taskId, unsigned int ttl = USE_TASK_TTL_VALUE);
  int cleanupExpiredTasksByTTL(unsigned int maxRemoveTasks = 1, unsigned int ttl = USE_TASK_TTL_VALUE);
  unsigned int cleanupExpiredTasksByTTLQueue( unsigned int maxRemoveItems, unsigned int minTerminateTime );
  void getExceededTTLTasks( unsigned int maxTaskNumber, std::vector<unsigned long long>& exceededTTLTasks,
                            unsigned int ttl = USE_TASK_TTL_VALUE) const;
  
private:
  Poco::SharedPtr<ReducingPartsCountersStorageIf> _reducingPartsCountersStorage;
  Poco::SharedPtr<ReduceAdditionsStorageIf> _reduceAdditionsStorage;
  Poco::SharedPtr<TaskReducersManagerFactoryIf> _taskReducersManagerFactory;
  Poco::SharedPtr<ReducingOutputMessageBuilderIf> _reducingOutputMessageBuilder;
  Poco::SharedPtr<SupportNotifierIf> _supportNotifier;
  ReducingTaskTTLManager _taskTTLManager;

  ReducingHandler();
  void createNewTaskReducersManager(HCE::types::MessageType reducerType);
};

}
}

#endif