Function RedisGearsPy_PyCallbackMapper

plugins/python/redisgears_python.c:5539–5601

}

static Record* RedisGearsPy_PyCallbackMapper(ExecutionCtx* rctx, Record *record, void* arg){
    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* oldObj = PyObjRecordGet(record);
    PyObjRecordSet(record, NULL); // pass ownership of oldObj to NULL
    PyTuple_SetItem(pArgs, 0, oldObj);
    if (RedisGears_ProfileEnabled()) {
        // start profiling
        PyObject_CallFunction(profileStartFunction, "O", sctx->profiler);
    }
    PyObject* newObj = PyObject_CallObject(callback, pArgs);
    if (RedisGears_ProfileEnabled()) {
        // start profiling
        PyObject_CallFunction(profileStopFunction, "O", sctx->profiler);
    }
    GearsPyDecRef(pArgs);

    PythonThreadCtx* ptctx = GetPythonThreadCtx();

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

    if(ptctx->pyfutureCreated && newObj == ptctx->pyfutureCreated){
        PyFuture* future = (PyFuture*)newObj;
        future->continueType = ContinueType_Default;
        GearsPyDecRef(newObj);
        RedisGears_FreeRecord(record);
        RedisGearsPy_Unlock(&pectx);
        return RedisGears_GetDummyRecord();
    }

    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(!newObj){
        fetchPyError(rctx);

        RedisGearsPy_Unlock(&pectx);
        RedisGears_FreeRecord(record);
        return NULL;
    }