Function continueFutureOnDone

plugins/python/redisgears_python.c:1432–1483

}

static void continueFutureOnDone(ExecutionPlan* ep, void* privateData){
    // create the results lists
    FlatExecutionPlan* fep = RedisGears_GetFep(ep);
    PythonSessionCtx* sctx = RedisGears_GetFlatExecutionPrivateDataFromFep(fep);
    RedisModule_Assert(sctx);

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

    PyObject* future = privateData;
    PyObject* l = PyList_New(0);
    PyObject* results = PyList_New(0);
    PyObject* errs = PyList_New(0);

    PyList_Append(l, results);
    PyList_Append(l, errs);

    for(size_t i = 0 ; i < RedisGears_GetRecordsLen(ep) ; ++i){
        Record* res = RedisGears_GetRecord(ep, i);
        RedisModule_Assert(RedisGears_RecordGetType(res) == pythonRecordType);
        PythonRecord* pyRec = (PythonRecord*)res;
        PyList_Append(results, pyRec->obj);
    }

    for(size_t i = 0 ; i < RedisGears_GetErrorsLen(ep) ; ++i){
        Record* err = RedisGears_GetError(ep, i);
        RedisModule_Assert(RedisGears_RecordGetType(err) == RedisGears_GetErrorRecordType());
        size_t len;
        const char* errStr = RedisGears_StringRecordGet(err, &len);
        PyObject* pyErr = PyUnicode_FromStringAndSize(errStr, len);
        PyList_Append(errs, pyErr);
        GearsPyDecRef(pyErr);
    }

    GearsPyDecRef(results);
    GearsPyDecRef(errs);

    PyObject* pArgs = PyTuple_New(2);
    PyTuple_SetItem(pArgs, 0, future);
    PyTuple_SetItem(pArgs, 1, l);
    PyObject* r = PyObject_CallObject(setFutureResultsFunction, pArgs);
    GearsPyDecRef(pArgs);
    if(!r){
        char* err = getPyError();
        RedisModule_Log(staticCtx, "warning", "Error happened when releasing execution future, error='%s'", err);
        RG_FREE(err);
    }else{
        GearsPyDecRef(r);
    }

    RedisGearsPy_Unlock(&pectx);
}

static void* runCreateStreamReaderArgs(const char* pattern, PyObject *kargs){
    const char* defaultFromIdStr = "0-0";