doSubmitMsg(const instanceConf_t *const __restrict__ inst,
	struct conn_wrkr_s *connWrkr, const uchar* msg, size_t len)
{
	smsg_t *pMsg;
	DEFiRet;

	assert(len <= s_iMaxLine);
	if (len == 0) {
		DBGPRINTF("discarding zero-sized message\n");
		FINALIZE;
	}

	CHKiRet(msgConstruct(&pMsg));
	MsgSetFlowControlType(pMsg, inst->flowControl
			            ? eFLOWCTL_LIGHT_DELAY : eFLOWCTL_NO_DELAY);
	if (inst->pInputName) {
		MsgSetInputName(pMsg, inst->pInputName);
	} else {
		MsgSetInputName(pMsg, pInputName);
	}
	MsgSetRawMsg(pMsg, (const char*)msg, len);
	MsgSetMSGoffs(pMsg, 0);	/* we do not have a header... */
	if (connWrkr->propRemoteAddr) {
		MsgSetRcvFromIP(pMsg, connWrkr->propRemoteAddr);
	}
	if (inst) {
		MsgSetRuleset(pMsg, inst->pBindRuleset);
	}
	// TODO: make these flags configurable.
	pMsg->msgFlags = NEEDS_PARSING | PARSE_HOSTNAME;

	if (inst->bAddMetadata) {
		CHKiRet(msgAddMetadataFromHttpHeader(pMsg, connWrkr));
		CHKiRet(msgAddMetadataFromHttpQueryParams(pMsg, connWrkr));
	}

	ratelimitAddMsg(inst->ratelimiter, &connWrkr->multiSub, pMsg);
	STATSCOUNTER_INC(statsCounter.ctrSubmitted, statsCounter.mutCtrSubmitted);
finalize_it:
	connWrkr->iMsg = 0;
	if (iRet != RS_RET_OK) {
		STATSCOUNTER_INC(statsCounter.ctrDiscarded, statsCounter.mutCtrDiscarded);
	}
	RETiRet;
}
doSubmitMsg(const instanceConf_t *const __restrict__ inst,
	struct conn_wrkr_s *connWrkr, const uchar* msg, size_t len)
{
	smsg_t *pMsg;
	DEFiRet;

	assert(len <= s_iMaxLine);
	if (len == 0) {
		DBGPRINTF("discarding zero-sized message\n");
		FINALIZE;
	}

	CHKiRet(msgConstruct(&pMsg));
	MsgSetFlowControlType(pMsg, inst->flowControl
			            ? eFLOWCTL_LIGHT_DELAY : eFLOWCTL_NO_DELAY);
	if (inst->pInputName) {
		MsgSetInputName(pMsg, inst->pInputName);
	} else {
		MsgSetInputName(pMsg, pInputName);
	}
	MsgSetRawMsg(pMsg, (const char*)msg, len);
	MsgSetMSGoffs(pMsg, 0);	 
	if (connWrkr->propRemoteAddr) {
		MsgSetRcvFromIP(pMsg, connWrkr->propRemoteAddr);
	}
	if (inst) {
		MsgSetRuleset(pMsg, inst->pBindRuleset);
	}
	 
	pMsg->msgFlags = NEEDS_PARSING | PARSE_HOSTNAME;

	if (inst->bAddMetadata) {
		CHKiRet(msgAddMetadataFromHttpHeader(pMsg, connWrkr));
		CHKiRet(msgAddMetadataFromHttpQueryParams(pMsg, connWrkr));
	}

	ratelimitAddMsg(inst->ratelimiter, &connWrkr->multiSub, pMsg);
	STATSCOUNTER_INC(statsCounter.ctrSubmitted, statsCounter.mutCtrSubmitted);
finalize_it:
	connWrkr->iMsg = 0;
	if (iRet != RS_RET_OK) {
		STATSCOUNTER_INC(statsCounter.ctrDiscarded, statsCounter.mutCtrDiscarded);
	}
	RETiRet;
}