Function PythonReader_Next

plugins/python/redisgears_python.c:6652–6694

}PythonReaderCtx;

static Record* PythonReader_Next(ExecutionCtx* rctx, void* ctx){
    PythonReaderCtx* pyCtx = ctx;
    if(pyCtx->isDone){
        return NULL;
    }

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

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

    PyObject* pyRecord = NULL;
    if(!pyCtx->generator){
        PyObject* pArgs = PyTuple_New(0);
        PyObject* callback = pyCtx->callback;
        pyRecord = PyObject_CallObject(callback, pArgs);
        GearsPyDecRef(pArgs);
        if(!pyRecord){
            fetchPyError(rctx);
            RedisGearsPy_Unlock(&pectx);
            pyCtx->isDone = true;
            return NULL;
        }
        if(PyGen_Check(pyRecord)) {
            pyCtx->generator = PyObject_GetIter(pyRecord);
            GearsPyDecRef(pyRecord);
            pyRecord = PyIter_Next(pyCtx->generator);
        }
    }else{
        pyRecord = PyIter_Next(pyCtx->generator);
    }
    if(!pyRecord){
        fetchPyError(rctx);
        RedisGearsPy_Unlock(&pectx);
        pyCtx->isDone = true;
        return NULL;
    }
    RedisGearsPy_Unlock(&pectx);
    Record* record = PyObjRecordCreate();
    PyObjRecordSet(record, pyRecord);
    return record;
}

static void PythonReader_Free(void* ctx){
    PythonReaderCtx* pyCtx = ctx;