cv::gapi::wip::IStreamSource::Ptr src;

virtual void halt() override {

src->halt();

}

virtual bool pull(cv::GRunArg &arg) override {

// FIXME: probably we can maintain a pool of (then) pre-allocated

// buffers to avoid runtime allocations.

// Pool size can be determined given the internal queue size.

cv::gapi::wip::Data newData;

if (!src->pull(newData)) {

return false;

}

arg = std::move(static_cast<cv::GRunArg&>(newData));

return true;

}

explicit VideoEmitter(const cv::GRunArg &arg) {

src = cv::util::get<cv::gapi::wip::IStreamSource::Ptr>(arg);

}

cv::GRunArg m_arg;

virtual void halt() override {

// Not used here, but in fact can be used.

}

virtual bool pull(cv::GRunArg &arg) override {

arg = const_cast<const cv::GRunArg&>(m_arg); // FIXME: variant workaround

return true;

}

explicit ConstEmitter(const cv::GRunArg &arg) : m_arg(arg) {

}

static const char *name() { return "StreamingDataQueue"; }

enum tag { DESYNC }; // Enum of 1 element: purely a syntax sugar

explicit DataQueue(std::size_t capacity) {

// Note: `ptr` is shared<SyncQueue>, while the `q` is a shared<Q>

auto ptr = std::make_shared<cv::gimpl::stream::SyncQueue>();

if (capacity != 0) {

ptr->set_capacity(capacity);

}

q = std::move(ptr);

}

explicit DataQueue(tag t)

: q(new cv::gimpl::stream::DesyncQueue()) {

GAPI_Assert(t == DESYNC);

}

// FIXME: ADE metadata requires types to be copiable

std::shared_ptr<cv::gimpl::stream::Q> q;

const ade::NodeHandle &obj)

{

ade::TypedGraph<DataQueue> qgr(g);

std::vector<cv::gimpl::stream::Q*> result;

for (auto &&out_eh : obj->outEdges())

{

result.push_back(qgr.metadata(out_eh).get<DataQueue>().q.get());

}

return result;

const ade::NodeHandle &obj)

{

ade::TypedGraph<DataQueue> qgr(g);

std::vector<cv::gimpl::stream::Q*> result;

for (auto &&in_eh : obj->inEdges())

{

result.push_back(qgr.metadata(in_eh).contains<DataQueue>()

? qgr.metadata(in_eh).get<DataQueue>().q.get()

: nullptr);

}

return result;

{

for (auto && it : ade::util::zip(ade::util::toRange(outputs),

ade::util::toRange(results)))

{

auto &out_obj = std::get<0>(it);

auto &res_obj = std::get<1>(it);

// FIXME: this conversion should be unified

using T = cv::GRunArgP;

switch (out_obj.index())

{

case T::index_of<cv::Mat*>():

{

auto out_mat_p = cv::util::get<cv::Mat*>(out_obj);

auto view = cv::util::get<cv::RMat>(res_obj).access(cv::RMat::Access::R);

*out_mat_p = cv::gimpl::asMat(view).clone();

} break;

case T::index_of<cv::RMat*>():

*cv::util::get<cv::RMat*>(out_obj) = std::move(cv::util::get<cv::RMat>(res_obj));

break;

case T::index_of<cv::Scalar*>():

*cv::util::get<cv::Scalar*>(out_obj) = std::move(cv::util::get<cv::Scalar>(res_obj));

break;

case T::index_of<cv::detail::VectorRef>():

cv::util::get<cv::detail::VectorRef>(out_obj).mov(cv::util::get<cv::detail::VectorRef>(res_obj));

break;

case T::index_of<cv::detail::OpaqueRef>():

cv::util::get<cv::detail::OpaqueRef>(out_obj).mov(cv::util::get<cv::detail::OpaqueRef>(res_obj));

break;

case T::index_of<cv::MediaFrame*>():

*cv::util::get<cv::MediaFrame*>(out_obj) = std::move(cv::util::get<cv::MediaFrame>(res_obj));

break;

default:

GAPI_Error("This value type is not supported!"); // ...maybe because of STANDALONE mode.

break;

}

}

{

namespace own = cv::gapi::own;

for (auto && it : ade::util::zip(ade::util::toRange(outputs),

ade::util::toRange(r.args),

ade::util::toRange(r.flags)))

{

auto &out_obj = std::get<0>(it);

auto &res_obj = std::get<1>(it);

bool available = std::get<2>(it);

using T = cv::GOptRunArgP;

#define HANDLE_CASE(Type) \

case T::index_of<O<Type>*>(): \

if (available) { \

*cv::util::get<O<Type>*>(out_obj) \

= cv::util::make_optional(std::move(cv::util::get<Type>(res_obj))); \

} else { \

cv::util::get<O<Type>*>(out_obj)->reset(); \

}

// FIXME: this conversion should be unified

switch (out_obj.index())

{

HANDLE_CASE(cv::Scalar); break;

HANDLE_CASE(cv::RMat); break;

HANDLE_CASE(cv::MediaFrame); break;

case T::index_of<O<cv::Mat>*>(): {

// Mat: special handling.

auto &mat_opt = *cv::util::get<O<cv::Mat>*>(out_obj);

if (available) {

auto q_map = cv::util::get<cv::RMat>(res_obj).access(cv::RMat::Access::R);

// FIXME: Copy! Maybe we could do some optimization for this case!

// e.g. don't handle RMat for last ilsand in the graph.

// It is not always possible though.

mat_opt = cv::util::make_optional(cv::gimpl::asMat(q_map).clone());

} else {

mat_opt.reset();

}

} break;

case T::index_of<cv::detail::OptionalVectorRef>(): {

// std::vector<>: special handling

auto &vec_opt = cv::util::get<cv::detail::OptionalVectorRef>(out_obj);

if (available) {

vec_opt.mov(cv::util::get<cv::detail::VectorRef>(res_obj));

} else {

vec_opt.reset();

}

} break;

case T::index_of<cv::detail::OptionalOpaqueRef>(): {

// std::vector<>: special handling

auto &opq_opt = cv::util::get<cv::detail::OptionalOpaqueRef>(out_obj);

if (available) {

opq_opt.mov(cv::util::get<cv::detail::OpaqueRef>(res_obj));

} else {

opq_opt.reset();

}

} break;

default:

// ...maybe because of STANDALONE mode.

GAPI_Error("This value type is not supported!");

break;

}

}

#undef HANDLE_CASE

{

bool m_finishing = false; // Set to true once a "soft" stop is received

std::vector<Cmd> m_cmd;

void rewindToStop(std::vector<Q*> &in_queues,

const std::size_t this_id);

cv::gimpl::StreamMsg getInputVector (std::vector<Q*> &in_queues,

cv::GRunArgs &in_constants);

using V = cv::util::variant<cv::GRunArgs, Stop, cv::gimpl::Exception>;

V getResultsVector(std::vector<Q*> &in_queues,

const std::vector<int> &in_mapping,

const std::size_t out_size);

const std::size_t this_id)

{

size_t expected_stop_count = std::count_if(in_queues.begin(), in_queues.end(), [] (const Q* ptr) {

return ptr != nullptr;

});

if (expected_stop_count > 0) {

// NB: it requires to subtract own queues id from total waiting queue count

// because it had got stop message before rewind was called

expected_stop_count--;

}

GAPI_LOG_DEBUG(nullptr, "id: " << this_id << ", queues count: " << in_queues.size() <<

", expected stop msg count: " << expected_stop_count);

size_t got_stop_count = 0;

while(got_stop_count < expected_stop_count) {

for (auto &&qit : ade::util::indexed(in_queues)) {

auto id2 = ade::util::index(qit);

auto &q2 = ade::util::value(qit);

if (this_id == id2) continue;

GAPI_LOG_DEBUG(nullptr, "drain next id: " << id2 <<

", stop count (" << got_stop_count << "/" <<

expected_stop_count << ")");

bool got_cmd = true;

while (q2 && got_cmd) {

Cmd cmd;

got_cmd = q2->try_pop(cmd);

if (got_cmd && cv::util::holds_alternative<Stop>(cmd)) {

got_stop_count ++;

GAPI_LOG_DEBUG(nullptr, "got stop from id: " << id2);

break;

}

}

}

}

GAPI_LOG_DEBUG(nullptr, "completed");

const std::size_t this_id)

{

::rewindToStop(in_queues, this_id);

cv::GRunArgs &in_constants)

{

// NB: Need to release resources from the previous step, to fetch new ones.

// On some systems it might be impossible to allocate new memory

// until the old one is released.

m_cmd.clear();

// NOTE: in order to maintain the GRunArg's underlying object

// lifetime, keep the whole cmd vector (of size == # of inputs)

// in memory.

m_cmd.resize(in_queues.size());

cv::GRunArgs isl_inputs(in_queues.size());

cv::optional<cv::gimpl::Exception> exception;

for (auto &&it : ade::util::indexed(in_queues))

{

auto id = ade::util::index(it);

auto &q = ade::util::value(it);

if (q == nullptr)

{

GAPI_Assert(!in_constants.empty());

// NULL queue means a graph-constant value (like a

// value-initialized scalar)

// It can also hold a constant value received with

// Stop::Kind::CNST message (see above).

isl_inputs[id] = in_constants[id];

continue;

}

q->pop(m_cmd[id]);

switch (m_cmd[id].index())

{

case Cmd::index_of<cv::GRunArg>():

isl_inputs[id] = cv::util::get<cv::GRunArg>(m_cmd[id]);

break;

case Cmd::index_of<Stop>():

{

const auto &stop = cv::util::get<Stop>(m_cmd[id]);

if (stop.kind == Stop::Kind::CNST)

{

// We've got a Stop signal from a const source,

// propagated as a result of real stream reaching its

// end. Sometimes these signals come earlier than

// real EOS Stops so are deprioritized -- just

// remember the Const value here and continue

// processing other queues. Set queue pointer to

// nullptr and update the const_val vector

// appropriately

m_finishing = true;

in_queues[id] = nullptr;

in_constants.resize(in_queues.size());

in_constants[id] = std::move(stop.cdata);

// NEXT time (on a next call to getInputVector()), the

// "q==nullptr" check above will be triggered, but now

// we need to make it manually:

isl_inputs[id] = in_constants[id];

}

else

{

GAPI_Assert(stop.kind == Stop::Kind::HARD);

rewindToStop(in_queues, id);

// After queues are read to the proper indicator,

// indicate end-of-stream

return cv::gimpl::StreamMsg{cv::gimpl::EndOfStream{}};

} // if(Cnst)

break;

}

case Cmd::index_of<cv::gimpl::Exception>():

{

exception =

cv::util::make_optional(cv::util::get<cv::gimpl::Exception>(m_cmd[id]));

break;

}

default:

GAPI_Error("Unsupported cmd type in getInputVector()");

}

} // for(in_queues)

if (exception.has_value()) {

return cv::gimpl::StreamMsg{exception.value()};

}

if (m_finishing)

{

// If the process is about to end (a soft Stop was received

// already) and an island has no other inputs than constant

// inputs, its queues may all become nullptrs. Indicate it as

// "no data".

if (ade::util::all_of(in_queues, [](Q *ptr){return ptr == nullptr;})) {

return cv::gimpl::StreamMsg{cv::gimpl::EndOfStream{}};

}

}

// A regular case - there is data to process

for (auto& arg : isl_inputs) {

if (arg.index() == cv::GRunArg::index_of<cv::Mat>()) {

arg = cv::GRunArg{ cv::make_rmat<cv::gimpl::RMatOnMat>(cv::util::get<cv::Mat>(arg))

, arg.meta

};

}

}

return cv::gimpl::StreamMsg{std::move(isl_inputs)};

const std::vector<int> &in_mapping,

const std::size_t out_size)

{

cv::GRunArgs out_results(out_size);

m_cmd.resize(out_size);

cv::optional<cv::gimpl::Exception> exception;

for (auto &&it : ade::util::indexed(in_queues))

{

auto ii = ade::util::index(it);

auto oi = in_mapping[ii];

auto &q = ade::util::value(it);

q->pop(m_cmd[oi]);

switch (m_cmd[oi].index()) {

case Cmd::index_of<cv::GRunArg>():

out_results[oi] = std::move(cv::util::get<cv::GRunArg>(m_cmd[oi]));

break;

case Cmd::index_of<Stop>():

// In theory, the CNST should never reach here.

// Collector thread never handles the inputs directly

// (collector's input queues are always produced by

// islands in the graph).

rewindToStop(in_queues, ii);

return QueueReader::V(Stop{});

case Cmd::index_of<cv::gimpl::Exception>():

exception =

cv::util::make_optional(cv::util::get<cv::gimpl::Exception>(m_cmd[oi]));

break;

default:

cv::util::throw_error(

std::logic_error("Unexpected cmd kind in getResultsVector"));

} // switch

} // for(in_queues)

if (exception.has_value()) {

return QueueReader::V(exception.value());

}

return QueueReader::V(out_results);

Q& in_queue,

std::vector<Q*> out_queues,

std::function<void()> cb_completion)

{

// Wait for the explicit Start command.

// ...or Stop command, this also happens.

Cmd cmd;

in_queue.pop(cmd);

GAPI_Assert( cv::util::holds_alternative<Start>(cmd)

|| cv::util::holds_alternative<Stop>(cmd));

if (cv::util::holds_alternative<Stop>(cmd))

{

for (auto &&oq : out_queues) {

oq->push(cmd);

}

return;

}

GAPI_ITT_STATIC_LOCAL_HANDLE(emitter_hndl, "emitter");

GAPI_ITT_STATIC_LOCAL_HANDLE(emitter_pull_hndl, "emitter_pull");

GAPI_ITT_STATIC_LOCAL_HANDLE(emitter_push_hndl, "emitter_push");

// Now start emitting the data from the source to the pipeline.

while (true)

{

GAPI_ITT_AUTO_TRACE_GUARD(emitter_hndl);

Cmd cancel;

if (in_queue.try_pop(cancel))

{

// if we just popped a cancellation command...

GAPI_Assert(cv::util::holds_alternative<Stop>(cancel));

// Broadcast it to the readers and quit.

for (auto &&oq : out_queues) oq->push(cancel);

return;

}

// Try to obtain next data chunk from the source

cv::GRunArg data;

bool result = false;

try {

result = [&](){

GAPI_ITT_AUTO_TRACE_GUARD(emitter_pull_hndl);

return emitter->pull(data);

}();

} catch (...) {

auto eptr = std::current_exception();

for (auto &&oq : out_queues)

{

oq->push(Cmd{cv::gimpl::Exception{eptr}});

}

// NB: Go to the next iteration.

continue;

}

if (result)

{

GAPI_ITT_AUTO_TRACE_GUARD(emitter_push_hndl);

// // On success, broadcast it to our readers

for (auto &&oq : out_queues)

{

// FIXME: FOR SOME REASON, oq->push(Cmd{data}) doesn't work!!

// empty mats are arrived to the receivers!

// There may be a fatal bug in our variant!

const auto tmp = data;

oq->push(Cmd{tmp});

}

}

else

{

// Otherwise, broadcast STOP message to our readers and quit.

// This usually means end-of-stream, so trigger a callback

for (auto &&oq : out_queues) oq->push(Cmd{Stop{}});

if (cb_completion) cb_completion();

return;

}

}

std::vector<std::vector<Q*>> out_queues) {

using timestamp_t = int64_t;

std::vector<bool> pop_nexts(in_queues.size());

std::vector<Cmd> cmds(in_queues.size());

GAPI_ITT_STATIC_LOCAL_HANDLE(sync_hndl, "sync_actor");

GAPI_ITT_STATIC_LOCAL_HANDLE(sync_pull_1_queue_hndl, "sync_actor_pull_from_1_queue");

GAPI_ITT_STATIC_LOCAL_HANDLE(sync_push_hndl, "sync_actor_push");

while (true) {

GAPI_ITT_AUTO_TRACE_GUARD(sync_hndl);

// pop_nexts indicates which queue still contains earlier timestamps and

// needs to be popped at least one more time.

// For each iteration (frame) we need to pull from each input queue at least once,

// so switch all to true when start processing new frame

for (auto&& p : pop_nexts) {

p = true;

}

timestamp_t max_ts = 0u;

// Iterate through all input queues, pop GRunArg's and compare timestamps.

// Continue pulling from queues whose timestamps are smaller.

// Finish when all timestamps are equal.

do {

for (auto&& it : ade::util::indexed(

ade::util::zip(pop_nexts, in_queues, cmds))) {

auto& val = ade::util::value(it);

auto& pop_next = std::get<0>(val);

if (!pop_next) {

continue;

}

auto& q = std::get<1>(val);

auto& cmd = std::get<2>(val);

{

GAPI_ITT_AUTO_TRACE_GUARD(sync_pull_1_queue_hndl);

q->pop(cmd);

}

if (cv::util::holds_alternative<Stop>(cmd)) {

// We got a stop command from one of the input queues.

// Rewind all input queues till Stop command,

// Push Stop command down the graph, finish the thread

rewindToStop(in_queues, ade::util::index(it));

for (auto &&oqs : out_queues) {

for (auto &&oq : oqs) {

oq->push(Cmd{Stop{}});

}

}

return;

}

// Extract the timestamp

auto& arg = cv::util::get<cv::GRunArg>(cmd);

auto ts = cv::util::any_cast<int64_t>(arg.meta[cv::gapi::streaming::meta_tag::timestamp]);

GAPI_Assert(ts >= 0u);

// TODO: this whole drop logic can be imported via compile args

// to give a user a way to customize it

if (ts < max_ts) {

// Continue popping from this queue

pop_next = true;

} else if (ts == max_ts) {

// Stop popping from this queue

pop_next = false;

} else if (ts > max_ts) {

// We got a timestamp which is greater than timestamps from other queues.

// It means that we need to reiterate through all the queues one more time

// (except the current one)

max_ts = ts;

for (auto&& p : pop_nexts) {

p = true;

}

pop_next = false;

}

}

} while (ade::util::any_of(pop_nexts, [](bool v){ return v; }));

// Finally we got all our inputs synchronized, push them further down the graph

{

GAPI_ITT_AUTO_TRACE_GUARD(sync_push_hndl);

for (auto &&it : ade::util::zip(out_queues, cmds)) {

for (auto &&q : std::get<0>(it)) {

q->push(std::get<1>(it));

}

}

}

}

{

QueueReader &qr;

std::vector<Q*> &in_queues; // FIXME: This can be part of QueueReader

cv::GRunArgs &in_constants; // FIXME: This can be part of QueueReader

cv::optional<cv::gimpl::StreamMsg> last_read_msg;

virtual cv::gimpl::StreamMsg try_get() override

{

// FIXME: This is not very usable at the moment!

return get();

}

public:

explicit StreamingInput(QueueReader &rdr,

std::vector<Q*> &inq,

cv::GRunArgs &inc,

const std::vector<cv::gimpl::RcDesc> &in_descs)

: qr(rdr), in_queues(inq), in_constants(inc)

{

set(in_descs);

}

const cv::gimpl::StreamMsg& read()

{

GAPI_ITT_STATIC_LOCAL_HANDLE(inputs_get_hndl, "StreamingInput::read");

GAPI_ITT_AUTO_TRACE_GUARD(inputs_get_hndl);

last_read_msg =

cv::optional<cv::gimpl::StreamMsg>(

qr.getInputVector(in_queues, in_constants));

return last_read_msg.value();

}

virtual cv::gimpl::StreamMsg get() override

{

GAPI_ITT_STATIC_LOCAL_HANDLE(inputs_get_hndl, "StreamingInput::get");

GAPI_ITT_AUTO_TRACE_GUARD(inputs_get_hndl);

if (!last_read_msg.has_value()) {

(void)read();

}

auto msg = std::move(last_read_msg.value());

last_read_msg = cv::optional<cv::gimpl::StreamMsg>();

return msg;

}

{

// These objects form an internal state of the StreamingOutput

struct Posting

{

using V = cv::util::variant<cv::GRunArg,

cv::gimpl::EndOfStream,

cv::gimpl::Exception>;

V data;

bool ready = false;

};

using PostingList = std::list<Posting>;

std::vector<PostingList> m_postings;

std::unordered_map< const void*

, std::pair<int, PostingList::iterator>

> m_postIdx;

std::size_t m_stops_sent = 0u;

// These objects are owned externally

const cv::GMetaArgs &m_metas;

std::vector< std::vector<Q*> > &m_out_queues;

std::shared_ptr<cv::gimpl::GIslandExecutable> m_island;

// NB: StreamingOutput have to be thread-safe.

// Now synchronization approach is quite poor and inefficient.

mutable std::mutex m_mutex;

// Allocate a new data object for output under idx

// Prepare this object for posting

virtual cv::GRunArgP get(int idx) override

{

GAPI_ITT_STATIC_LOCAL_HANDLE(outputs_get_hndl, "StreamingOutput::get (alloc)");

GAPI_ITT_AUTO_TRACE_GUARD(outputs_get_hndl);

std::lock_guard<std::mutex> lock{m_mutex};

using MatType = cv::Mat;

using SclType = cv::Scalar;

// Allocate a new posting first, then bind this GRunArgP to this item

auto iter = m_postings[idx].insert(m_postings[idx].end(), Posting{});

const auto r = desc()[idx];

cv::GRunArg& out_arg = cv::util::get<cv::GRunArg>(iter->data);

cv::GRunArgP ret_val;

switch (r.shape) {

// Allocate a data object based on its shape & meta, and put it into our vectors.

// Yes, first we put a cv::Mat GRunArg, and then specify _THAT_

// pointer as an output parameter - to make sure that after island completes,

// our GRunArg still has the right (up-to-date) value.

// Same applies to other types.

// FIXME: This is absolutely ugly but seem to work perfectly for its purpose.

case cv::GShape::GMAT:

{

auto desc = cv::util::get<cv::GMatDesc>(m_metas[idx]);

if (m_island->allocatesOutputs())

{

out_arg = cv::GRunArg(m_island->allocate(desc));

}

else

{

MatType newMat;

cv::gimpl::createMat(desc, newMat);

auto rmat = cv::make_rmat<cv::gimpl::RMatOnMat>(newMat);

out_arg = cv::GRunArg(std::move(rmat));

}

ret_val = cv::GRunArgP(&cv::util::get<cv::RMat>(out_arg));

}

break;

case cv::GShape::GSCALAR:

{

SclType newScl;

out_arg = cv::GRunArg(std::move(newScl));

ret_val = cv::GRunArgP(&cv::util::get<SclType>(out_arg));

}

break;

case cv::GShape::GARRAY:

{

cv::detail::VectorRef newVec;

cv::util::get<cv::detail::ConstructVec>(r.ctor)(newVec);

out_arg = cv::GRunArg(std::move(newVec));

// VectorRef is implicitly shared so no pointer is taken here

// FIXME: that variant MOVE problem again

const auto &rr = cv::util::get<cv::detail::VectorRef>(out_arg);

ret_val = cv::GRunArgP(rr);

}

break;

case cv::GShape::GOPAQUE:

{

cv::detail::OpaqueRef newOpaque;

cv::util::get<cv::detail::ConstructOpaque>(r.ctor)(newOpaque);

out_arg = cv::GRunArg(std::move(newOpaque));

// OpaqueRef is implicitly shared so no pointer is taken here

// FIXME: that variant MOVE problem again

const auto &rr = cv::util::get<cv::detail::OpaqueRef>(out_arg);

ret_val = cv::GRunArgP(rr);

}

break;

case cv::GShape::GFRAME:

{

cv::MediaFrame frame;

out_arg = cv::GRunArg(std::move(frame));

ret_val = cv::GRunArgP(&cv::util::get<cv::MediaFrame>(out_arg));

}

break;

default:

cv::util::throw_error(std::logic_error("Unsupported GShape"));

}

m_postIdx[cv::gimpl::proto::ptr(ret_val)] = std::make_pair(idx, iter);

return ret_val;

}

virtual void post(cv::GRunArgP&& argp, const std::exception_ptr& exptr) override

{

GAPI_ITT_STATIC_LOCAL_HANDLE(outputs_post_hndl, "StreamingOutput::post");

GAPI_ITT_AUTO_TRACE_GUARD(outputs_post_hndl);

std::lock_guard<std::mutex> lock{m_mutex};

// Mark the output ready for posting. If it is the first in the line,

// actually post it and all its successors which are ready for posting too.

auto it = m_postIdx.find(cv::gimpl::proto::ptr(argp));

GAPI_Assert(it != m_postIdx.end());

const int out_idx = it->second.first;

const auto out_iter = it->second.second;

out_iter->ready = true;

if (exptr) {

out_iter->data = cv::gimpl::Exception{exptr};

}

m_postIdx.erase(it); // Drop the link from the cache anyway

if (out_iter != m_postings[out_idx].begin())

{

return; // There are some pending postings in the beginning, return

}

GAPI_Assert(out_iter == m_postings[out_idx].begin());

auto post_iter = m_postings[out_idx].begin();

while (post_iter != m_postings[out_idx].end() && post_iter->ready == true)

{

Cmd cmd;

switch (post_iter->data.index())

{

case Posting::V::index_of<cv::GRunArg>():

cmd = Cmd{cv::util::get<cv::GRunArg>(post_iter->data)};

break;

case Posting::V::index_of<cv::gimpl::Exception>():

cmd = Cmd{cv::util::get<cv::gimpl::Exception>(post_iter->data)};

break;

case Posting::V::index_of<cv::gimpl::EndOfStream>():

cmd = Cmd{Stop{}};

m_stops_sent++;

break;

default:

GAPI_Error("Unreachable code");

}

for (auto &&q : m_out_queues[out_idx])

{

q->push(cmd);

}

post_iter = m_postings[out_idx].erase(post_iter);

}

}

virtual void post(cv::gimpl::EndOfStream&&) override

{

std::lock_guard<std::mutex> lock{m_mutex};

// If the posting list is empty, just broadcast the stop message.

// If it is not, enqueue the Stop message in the postings list.

for (auto &&it : ade::util::indexed(m_postings))

{

const auto idx = ade::util::index(it);

auto &lst = ade::util::value(it);

if (lst.empty())

{

for (auto &&q : m_out_queues[idx])

{

q->push(Cmd(Stop{}));

}

m_stops_sent++;

}

else

{

Posting p;

p.data = Posting::V{cv::gimpl::EndOfStream{}};

p.ready = true;

lst.push_back(std::move(p)); // FIXME: For some reason {}-ctor didn't work here

}

}

}

void meta(const cv::GRunArgP &out, const cv::GRunArg::Meta &m) override

{

std::lock_guard<std::mutex> lock{m_mutex};

const auto it = m_postIdx.find(cv::gimpl::proto::ptr(out));

GAPI_Assert(it != m_postIdx.end());

const auto out_iter = it->second.second;

cv::util::get<cv::GRunArg>(out_iter->data).meta = m;

}

explicit StreamingOutput(const cv::GMetaArgs &metas,

std::vector< std::vector<Q*> > &out_queues,

const std::vector<cv::gimpl::RcDesc> &out_descs,

std::shared_ptr<cv::gimpl::GIslandExecutable> island)

: m_metas(metas)

, m_out_queues(out_queues)

, m_island(island)

{

set(out_descs);

m_postings.resize(out_descs.size());

}

bool done() const

{

std::lock_guard<std::mutex> lock{m_mutex};

// The streaming actor work is considered DONE for this stream

// when it posted/resent all STOP messages to all its outputs.

return m_stops_sent == desc().size();

}

virtual void post(cv::gimpl::Exception&& error) override