Method put

edrav2/iprj/libcore/src/queue.cpp:232–281

//
//
void Queue::put(const Variant& vData)
{
	// Apply filter
	if (m_pFilter && m_pFilter->filterElem(vData))
	{
		++m_nFilteredCount;
		return;
	}

	// Simple mode processing
	if (m_nBatchSize == 0)
	{
		putIntoData(vData, true);
		return;
	}

	// Batch mode processing
	{
		std::unique_lock sync(m_mtxData);
		if (!testFlag(m_eMode, QueueMode::Put))
			error::OperationDeclined().throwException();
	}

	std::scoped_lock lock(m_mtxBatch);

	m_vBatch.push_back(vData);
	auto nSize = m_vBatch.getSize();
	if (nSize >= m_nBatchSize)
	{
		if (m_pBatchTimer)
			m_pBatchTimer = nullptr;
		putIntoData(m_vBatch, true);
		m_vBatch = Sequence();
	}
	else if (m_nBatchTimeout != 0)
	{
		if (m_pBatchTimer == nullptr)
		{
			// avoid circular pointing timer <-> this
			ObjWeakPtr<Queue> weakPtrToThis(getPtrFromThis(this));
			m_pBatchTimer = runWithDelay(m_nBatchTimeout, [weakPtrToThis]()
			{
				auto pThis = weakPtrToThis.lock();
				if (pThis == nullptr)
					return false;
				return pThis->flushBatch();
			});
		}
	}
}

//
//