Function RedisGearsPy_PyCallbackForEach

plugins/python/redisgears_python.c:5308–5375

}

int RedisGearsPy_PyCallbackForEach(ExecutionCtx* rctx, Record *record, void* arg){
    // Call Python/C API functions...
    RedisModule_Assert(RedisGears_RecordGetType(record) == pythonRecordType);

    PythonSessionCtx* sctx = RedisGears_GetFlatExecutionPrivateData(rctx);
    RedisModule_Assert(sctx);

    PythonExecutionCtx pectx = PythonExecutionCtx_New(sctx, rctx);
    RedisGearsPy_Lock(&pectx);

    PyObject* pArgs = PyTuple_New(1);
    PyObject* callback = arg;
    PyObject* obj = PyObjRecordGet(record);
    Py_INCREF(obj);
    PyTuple_SetItem(pArgs, 0, obj);
    if (RedisGears_ProfileEnabled()) {
        // start profiling
        PyObject_CallFunction(profileStartFunction, "O", sctx->profiler);
    }
    PyObject* ret = PyObject_CallObject(callback, pArgs);
    if (RedisGears_ProfileEnabled()) {
        // stop profiling
        PyObject_CallFunction(profileStopFunction, "O", sctx->profiler);
    }
    GearsPyDecRef(pArgs);

    PythonThreadCtx* ptctx = GetPythonThreadCtx();

    if(ret && PyCoro_CheckExact(ret)){
        // object is a coroutine, we need to hold and pass it to the event loop.
        ret = RedisGearsPy_PyCallbackHandleCoroutine(rctx, ret, ptctx);
    }

    if(ptctx->pyfutureCreated && ret == ptctx->pyfutureCreated){
        PyFuture* future = (PyFuture*)ret;
        future->continueType = ContinueType_Foreach;
        // no need to free the original record because the async record took the ownership on it

        // we need to free the future as we do not return it
        GearsPyDecRef(ret);
        RedisGearsPy_Unlock(&pectx);
        return RedisGears_StepHold;
    }

    if(ptctx->pyfutureCreated){
        /**
         * Async record created but not returned lets disscard it
         */
        PyFuture* f = (PyFuture*)ptctx->pyfutureCreated;
        RedisGears_AsyncRecordContinue(f->asyncRecord, RedisGears_GetDummyRecord());
        f->asyncRecord = NULL;
    }

    if(!ret){
        fetchPyError(rctx);

        RedisGearsPy_Unlock(&pectx);
        return RedisGears_StepSuccess;