Function RedisGearsPy_PyCallbackReducer

plugins/python/redisgears_python.c:5807–5860

}

static Record* RedisGearsPy_PyCallbackReducer(ExecutionCtx* rctx, char* key, size_t keyLen, Record *records, void* arg){
    RedisModule_Assert(RedisGears_RecordGetType(records) == RedisGears_GetListRecordType());

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

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

    PyObject* obj = PyList_New(0);
    for(size_t i = 0 ; i < RedisGears_ListRecordLen(records) ; ++i){
        Record* r = RedisGears_ListRecordGet(records, i);
        RedisModule_Assert(RedisGears_RecordGetType(r) == pythonRecordType);
        PyObject* element = PyObjRecordGet(r);
        PyList_Append(obj, element);
    }
    PyObject* reducer = arg;
    PyObject* pArgs = PyTuple_New(2);
    PyObject* keyPyObj = PyUnicode_FromString(key);
    PyTuple_SetItem(pArgs, 0, keyPyObj);
    PyTuple_SetItem(pArgs, 1, obj);
    if (RedisGears_ProfileEnabled()) {
        // start profiling
        PyObject_CallFunction(profileStartFunction, "O", sctx->profiler);
    }
    PyObject* ret = PyObject_CallObject(reducer, pArgs);
    if (RedisGears_ProfileEnabled()) {
        // start profiling
        PyObject_CallFunction(profileStopFunction, "O", sctx->profiler);
    }
    GearsPyDecRef(pArgs);

    if(ret && PyCoro_CheckExact(ret)){
        GearsPyDecRef(ret);
        RedisGears_SetError(rctx, RG_STRDUP("coroutine are not allow on reduce"));
        RedisGearsPy_Unlock(&pectx);
        RedisGears_FreeRecord(records);
        return NULL;
    }

    if(!ret){
        fetchPyError(rctx);

        RedisGearsPy_Unlock(&pectx);
        RedisGears_FreeRecord(records);
        return NULL;
    }
    Record* retRecord = PyObjRecordCreate();
    PyObjRecordSet(retRecord, ret);

    RedisGearsPy_Unlock(&pectx);
    RedisGears_FreeRecord(records);
    return retRecord;
}

static Record* RedisGearsPy_ToPyRecordMapperInternal(ExecutionCtx* rctx, Record *record, void* arg){
    Record* res = PyObjRecordCreate();