#include <EventStreamHttpReader.h>

Inheritance diagram for edm::EventStreamHttpReader:

Public Member Functions
	EventStreamHttpReader (edm::ParameterSet const &, edm::InputSourceDescription const &)
virtual EventPrincipal *	read ()
virtual	~EventStreamHttpReader ()
Private Member Functions
void	initializeDQMStore ()
void	readHeader ()
Private Attributes
const std::string	consumerName_
DQMStore *	dqmStore_
bool	dqmStoreAvailabiltyChecked_
const bool	dropOldLumisectionEvents_
stor::EventServerProxy < stor::EventConsumerRegistrationInfo >	eventServerProxy_
unsigned int	lastLS_
unsigned int	totalDroppedEvents_

Detailed Description

Input source for event consumers that will get events from the Storage Manager Event Server. This does uses a HTTP get using the CURL library. The Storage Manager Event Server responses with a binary octet-stream. The product registry is also obtained through a HTTP get.

There is currently no test of the product registry against the consumer client product registry within the code. It should already be done if this was inherenting from the standard framework input source. Currently we inherit from InputSource.

Author:: mommsen

Revision:: 1.29

Date:: 2011/09/02 08:13:31

Definition at line 34 of file EventStreamHttpReader.h.

Constructor & Destructor Documentation

edm::EventStreamHttpReader::EventStreamHttpReader	(	edm::ParameterSet const &	pset,
		edm::InputSourceDescription const &	desc
	)

Definition at line 18 of file EventStreamHttpReader.cc.

References edm::ParameterSet::getUntrackedParameter(), and xuti::readHeader().

   :
  StreamerInputSource(pset, desc),
  eventServerProxy_(pset),
  dqmStore_(0),
  dqmStoreAvailabiltyChecked_(false),
  dropOldLumisectionEvents_(pset.getUntrackedParameter<bool>("dropOldLumisectionEvents", false)),
  consumerName_(pset.getUntrackedParameter<std::string>("consumerName")),
  totalDroppedEvents_(0),
  lastLS_(0)
  {
    // Default in StreamerInputSource is 'false'
    inputFileTransitionsEachEvent_ =
      pset.getUntrackedParameter<bool>("inputFileTransitionsEachEvent", true);

    readHeader();
  }

virtual edm::EventStreamHttpReader::~EventStreamHttpReader ( ) [inline, virtual]

Definition at line 42 of file EventStreamHttpReader.h.

{};

Member Function Documentation

void edm::EventStreamHttpReader::initializeDQMStore ( ) [private]

Definition at line 95 of file EventStreamHttpReader.cc.

References dqmStore_, dqmStoreAvailabiltyChecked_, alignCSCRings::e, and cmsCodeRules::cppFunctionSkipper::operator.

Referenced by read().

  {
    if ( ! dqmStoreAvailabiltyChecked_ )
    {   
      try
      {
        dqmStore_ = edm::Service<DQMStore>().operator->();
      }
      catch (cms::Exception& e)
      {
        edm::LogInfo("EventStreamHttpReader")
          << "Service DQMStore not defined. Will not record the number of dropped events.";
      }
      dqmStoreAvailabiltyChecked_ = true;
    }
  }

EventPrincipal * edm::EventStreamHttpReader::read ( ) [virtual]

Implements edm::StreamerInputSource.

Definition at line 39 of file EventStreamHttpReader.cc.

References DQMStore::bookInt(), HeaderView::code(), consumerName_, AlCaHLTBitMon_QueryRunRegistry::data, edm::StreamerInputSource::deserializeEvent(), Header::DONE, dqmStore_, dropOldLumisectionEvents_, EventMsgView::droppedEventsCount(), eventServerProxy_, MonitorElement::Fill(), DQMStore::get(), stor::EventServerProxy< RegInfo >::getOneEvent(), initializeDQMStore(), lastLS_, EventMsgView::lumi(), DQMStore::setCurrentFolder(), edm::StreamerInputSource::setEndRun(), and totalDroppedEvents_.

  {
    initializeDQMStore();

    stor::CurlInterface::Content data;
    unsigned int currentLS(0);
    unsigned int droppedEvents(0);
    
    do
    {
      eventServerProxy_.getOneEvent(data);
      if ( data.empty() ) return 0;
      
      HeaderView hdrView(&data[0]);
      if (hdrView.code() == Header::DONE)
      {
        setEndRun();
        return 0;
      }
      
      EventMsgView eventView(&data[0]);
      currentLS = eventView.lumi();
      droppedEvents += eventView.droppedEventsCount();
    }
    while (
      dropOldLumisectionEvents_ &&
      lastLS_ > currentLS
    );
    
    lastLS_ = currentLS;

    if (dqmStore_)
    {
      MonitorElement* me = dqmStore_->get("SM_SMPS_Stats/droppedEventsCount_" + consumerName_ );
      if (!me){
        dqmStore_->setCurrentFolder("SM_SMPS_Stats");
        me = dqmStore_->bookInt("droppedEventsCount_" + consumerName_ );
      }
      totalDroppedEvents_ += droppedEvents;
      me->Fill(totalDroppedEvents_);
    }

    return deserializeEvent(EventMsgView(&data[0]));
  }

void edm::EventStreamHttpReader::readHeader ( ) [private]

Definition at line 85 of file EventStreamHttpReader.cc.

References AlCaHLTBitMon_QueryRunRegistry::data, edm::StreamerInputSource::deserializeAndMergeWithRegistry(), eventServerProxy_, and stor::EventServerProxy< RegInfo >::getInitMsg().

  {
    stor::CurlInterface::Content data;
    
    eventServerProxy_.getInitMsg(data);
    InitMsgView initView(&data[0]);
    deserializeAndMergeWithRegistry(initView);
  }