Integration/plugins/AcquireIntStreamProducer.cc

0001 #include "DataFormats/Common/interface/Handle.h"
0002 #include "DataFormats/TestObjects/interface/ToyProducts.h"
0003 #include "FWCore/Concurrency/interface/WaitingTaskWithArenaHolder.h"
0004 #include "FWCore/Framework/interface/Event.h"
0005 #include "FWCore/Framework/interface/stream/EDProducer.h"
0006 #include "FWCore/Framework/interface/MakerMacros.h"
0007 #include "WaitingService.h"
0008 #include "FWCore/ParameterSet/interface/ParameterSet.h"
0009 #include "FWCore/ServiceRegistry/interface/Service.h"
0010 #include "FWCore/Utilities/interface/EDGetToken.h"
0011 #include "FWCore/Utilities/interface/InputTag.h"
0012
0013 #include <memory>
0014 #include <vector>
0015
0016 namespace edm {
0017   class EventSetup;
0018 }
0019
0020 namespace edmtest {
0021
0022   class AcquireIntStreamProducer : public edm::stream::EDProducer<edm::ExternalWork> {
0023   public:
0024     explicit AcquireIntStreamProducer(edm::ParameterSet const& pset);
0025     ~AcquireIntStreamProducer() override;
0026     void acquire(edm::Event const&, edm::EventSetup const&, edm::WaitingTaskWithArenaHolder) override;
0027     void produce(edm::Event&, edm::EventSetup const&) override;
0028
0029   private:
0030     std::vector<edm::EDGetTokenT<IntProduct>> m_getTokens;
0031     edm::EDGetTokenT<IntProduct> m_tokenForProduce;
0032     test_acquire::Token m_token;
0033   };
0034
0035   AcquireIntStreamProducer::AcquireIntStreamProducer(edm::ParameterSet const& pset)
0036       : m_token{edm::Service<test_acquire::WaitingService>()->getToken()} {
0037     for (auto const& tag : pset.getParameter<std::vector<edm::InputTag>>("tags")) {
0038       m_getTokens.emplace_back(consumes<IntProduct>(tag));
0039     }
0040     m_tokenForProduce = consumes<IntProduct>(pset.getParameter<edm::InputTag>("produceTag"));
0041     produces<IntProduct>();
0042   }
0043
0044   AcquireIntStreamProducer::~AcquireIntStreamProducer() {}
0045
0046   void AcquireIntStreamProducer::acquire(edm::Event const& event,
0047                                          edm::EventSetup const&,
0048                                          edm::WaitingTaskWithArenaHolder holder) {
0049     test_acquire::Cache* cache = edm::Service<test_acquire::WaitingService>()->getCache(m_token);
0050     cache->retrieved().clear();
0051     cache->processed().clear();
0052
0053     for (auto const& token : m_getTokens) {
0054       cache->retrieved().push_back(event.get(token).value);
0055     }
0056
0057     edm::Service<test_acquire::WaitingService>()->requestValuesAsync(
0058         m_token, &cache->retrieved(), &cache->processed(), holder);
0059   }
0060
0061   void AcquireIntStreamProducer::produce(edm::Event& event, edm::EventSetup const&) {
0062     int sum = 0;
0063     test_acquire::Cache* cache = edm::Service<test_acquire::WaitingService>()->getCache(m_token);
0064     for (auto v : cache->processed()) {
0065       sum += v;
0066     }
0067     event.put(std::make_unique<IntProduct>(sum));
0068
0069     // This part is here only for the Parentage test.
0070     (void)event.get(m_tokenForProduce);
0071   }
0072 }  // namespace edmtest
0073 using edmtest::AcquireIntStreamProducer;
0074 DEFINE_FWK_MODULE(AcquireIntStreamProducer);