view release on metacpan or  search on metacpan

build/lib/src/result_stream.c  view on Meta::CPAN

struct run_result_stream
{
    neo4j_result_stream_t _result_stream;

    neo4j_connection_t *connection;
    neo4j_job_t job;
    neo4j_logger_t *logger;
    neo4j_memory_allocator_t *allocator;
    neo4j_mpool_t mpool;
    neo4j_mpool_t record_mpool;
    unsigned int refcount;
    unsigned int starting;
    unsigned int streaming;
    int statement_type;
    struct neo4j_statement_plan *statement_plan;
    struct neo4j_update_counts update_counts;
    unsigned long long available_after;
    unsigned long long consumed_after;
    int failure;
    struct neo4j_failure_details failure_details;
    unsigned int nfields;
    const char *const *fields;
    result_record_t *records;
    result_record_t *records_tail;
    unsigned long long records_depth;
    result_record_t *last_fetched;
    unsigned long long nrecords;
    unsigned int awaiting_records;
};


static run_result_stream_t *run_rs_open(neo4j_connection_t *connection);
static int run_rs_check_failure(neo4j_result_stream_t *self);
static const char *run_rs_error_code(neo4j_result_stream_t *self);
static const char *run_rs_error_message(neo4j_result_stream_t *self);
const struct neo4j_failure_details *run_rs_failure_details(
        neo4j_result_stream_t *results);
static unsigned int run_rs_nfields(neo4j_result_stream_t *results);
static const char *run_rs_fieldname(neo4j_result_stream_t *self,
        unsigned int index);
static neo4j_result_t *run_rs_fetch_next(neo4j_result_stream_t *self);
static neo4j_result_t *run_rs_peek(neo4j_result_stream_t *self,
        unsigned int depth);
static unsigned long long run_rs_count(neo4j_result_stream_t *self);
static unsigned long long run_rs_available_after(neo4j_result_stream_t *self);
static unsigned long long run_rs_consumed_after(neo4j_result_stream_t *self);
static int run_rs_statement_type(neo4j_result_stream_t *self);
static struct neo4j_statement_plan *run_rs_statement_plan(
        neo4j_result_stream_t *self);
static struct neo4j_update_counts run_rs_update_counts(
        neo4j_result_stream_t *self);
static int run_rs_close(neo4j_result_stream_t *self);

static neo4j_value_t run_result_field(const neo4j_result_t *self,
        unsigned int index);
static neo4j_result_t *run_result_retain(neo4j_result_t *self);
static void run_result_release(neo4j_result_t *self);

static void abort_job(neo4j_job_t *job, int err);
static int run_callback(void *cdata, neo4j_message_type_t type,
        const neo4j_value_t *argv, uint16_t argc);
static int pull_all_callback(void *cdata, neo4j_message_type_t type,
        const neo4j_value_t *argv, uint16_t argc);
static int discard_all_callback(void *cdata, neo4j_message_type_t type,
        const neo4j_value_t *argv, uint16_t argc);
static int stream_end(run_result_stream_t *results, neo4j_message_type_t type,
        const char *src_message_type, const neo4j_value_t *argv, uint16_t argc);
static int await(run_result_stream_t *results, const unsigned int *condition);
static int append_result(run_result_stream_t *results,
        const neo4j_value_t *argv, uint16_t argc);
void result_record_release(result_record_t *record);
static int set_eval_failure(run_result_stream_t *results,
        const char *src_message_type, const neo4j_value_t *argv, uint16_t argc);
static void set_failure(run_result_stream_t *results, int error);

int neo4j_set_extra(neo4j_value_t map)
{
  REQUIRE(neo4j_type(map) == NEO4J_MAP || neo4j_is_null(map), -1);
  unsigned int i;
  const struct neo4j_map *p = (const struct neo4j_map *)↦
  for (i=0;i<neo4j_map_size(map);i++)
    {
      g_extra_ents[i] = p->entries[i];
    }
  g_extra_map = neo4j_map(g_extra_ents,i);
  return 0;
}

neo4j_result_stream_t *neo4j_run_in_db(neo4j_connection_t *connection,
        const char *statement, neo4j_value_t params, const char *dbname)
{
   REQUIRE(connection != NULL, NULL);
   REQUIRE(statement != NULL, NULL);
   REQUIRE(neo4j_type(params) == NEO4J_MAP || neo4j_is_null(params), NULL);
   REQUIRE(dbname != NULL, NULL);

   if (connection->version < 4)
     {
       errno = NEO4J_FEATURE_UNAVAILABLE;
       char buf[128];
       sprintf(buf,"named dbs not available in protocol version %d", connection->version);
       neo4j_log_error_errno(connection->logger, (const char*) buf);
       return NULL;
     }
   g_extra_ents[0] = neo4j_map_entry("db", neo4j_string(dbname));
   g_extra_map = neo4j_map(g_extra_ents,1);
   neo4j_result_stream_t *rs = neo4j_run(connection, statement, params);
   g_extra_map = neo4j_null;
   return rs;
}

neo4j_result_stream_t *neo4j_run(neo4j_connection_t *connection,
        const char *statement, neo4j_value_t params)
{
    REQUIRE(connection != NULL, NULL);
    REQUIRE(statement != NULL, NULL);
    REQUIRE(neo4j_type(params) == NEO4J_MAP || neo4j_is_null(params), NULL);
    REQUIRE(neo4j_type(g_extra_map) == NEO4J_MAP || neo4j_is_null(g_extra_map), NULL);
    run_result_stream_t *results = run_rs_open(connection);
    if (results == NULL)
    {
        return NULL;
    }

build/lib/src/result_stream.c  view on Meta::CPAN

        results->records = next;
    }

    neo4j_statement_plan_release(results->statement_plan);
    results->statement_plan = NULL;
    neo4j_logger_release(results->logger);
    results->logger = NULL;
    neo4j_mpool_drain(&(results->record_mpool));
    neo4j_mpool_drain(&(results->mpool));
    neo4j_free(results->allocator, results);
    return err;
}


neo4j_value_t run_result_field(const neo4j_result_t *self,
        unsigned int index)
{
    const result_record_t *record = container_of(self,
            result_record_t, _result);
    REQUIRE(record != NULL, neo4j_null);
    return neo4j_list_get(record->list, index);
}


neo4j_result_t *run_result_retain(neo4j_result_t *self)
{
    result_record_t *record = container_of(self,
            result_record_t, _result);
    REQUIRE(record != NULL, NULL);
    (record->refcount)++;
    return self;
}


void run_result_release(neo4j_result_t *self)
{
    result_record_t *record = container_of(self,
            result_record_t, _result);
    result_record_release(record);
}


void abort_job(neo4j_job_t *job, int err)
{
    run_result_stream_t *results = container_of(job,
            run_result_stream_t, job);
    if (results == NULL || results->connection == NULL)
    {
        return;
    }

    job->next = NULL;
    results->connection = NULL;
    if (results->streaming && results->failure == 0)
    {
        set_failure(results, err);
    }
}


int run_callback(void *cdata, neo4j_message_type_t type,
        const neo4j_value_t *argv, uint16_t argc)
{
    assert(cdata != NULL);
    assert(argc == 0 || argv != NULL);
    run_result_stream_t *results = (run_result_stream_t *)cdata;
    neo4j_logger_t *logger = results->logger;
    neo4j_connection_t *connection = results->connection;

    results->starting = false;
    --(results->refcount);

    if (type == NULL || connection == NULL)
    {
        return 0;
    }

#ifndef NEOCLIENT_BUILD    
    if (type == NEO4J_FAILURE_MESSAGE)
#else
    if ( MESSAGE_TYPE_IS(type,FAILURE) )
#endif	
    {
        return set_eval_failure(results, "RUN", argv, argc);
    }
    
#ifndef NEOCLIENT_BUILD    
    if (type == NEO4J_IGNORED_MESSAGE)
#else
    if ( MESSAGE_TYPE_IS(type,IGNORED) )
#endif
    {
        if (results->failure == 0)
        {
            set_failure(results, NEO4J_STATEMENT_PREVIOUS_FAILURE);
        }
        return 0;
    }

    char description[128];
    snprintf(description, sizeof(description), "%s in %p (response to RUN)",
            neo4j_message_type_str(type), (void *)connection);

#ifndef NEOCLIENT_BUILD
    if (type != NEO4J_SUCCESS_MESSAGE)
#else
    if ( !MESSAGE_TYPE_IS(type,SUCCESS) )
#endif
    {
        neo4j_log_error(logger, "Unexpected %s", description);
        set_failure(results, errno = EPROTO);
        return -1;
    }

    const neo4j_value_t *metadata = neo4j_validate_metadata(argv, argc,
            description, logger);
    if (metadata == NULL)
    {
        set_failure(results, errno);
        return -1;
    }

    if (neo4j_log_is_enabled(connection->logger, NEO4J_LOG_TRACE))
    {
        neo4j_metadata_log(logger, NEO4J_LOG_TRACE, description, *metadata);
    }

    if (neo4j_meta_fieldnames(&(results->fields), &(results->nfields),
                *metadata, &(results->mpool), description, logger))
    {
        set_failure(results, errno);
        return -1;
    }

    long long available_after =
            neo4j_meta_result_available_after(*metadata, description, logger);
    if (available_after < 0)
    {
        set_failure(results, errno);
        return -1;
    }
    results->available_after = available_after;

    return 0;
}


int pull_all_callback(void *cdata, neo4j_message_type_t type,
        const neo4j_value_t *argv, uint16_t argc)
{
    assert(cdata != NULL);
    assert(argc == 0 || argv != NULL);
    run_result_stream_t *results = (run_result_stream_t *)cdata;

#ifndef NEOCLIENT_BUILD
    if (type == NEO4J_RECORD_MESSAGE)
#else
    if ( MESSAGE_TYPE_IS(type,RECORD) )
#endif
    {
        if (append_result(results, argv, argc))
        {
            neo4j_log_trace_errno(results->logger, "append_result failed");
            set_failure(results, errno);
            return -1;
        }
        return 1;
    }

    --(results->refcount);
    results->streaming = false;

    // not a record, so keep this memory along with the result stream
    if (neo4j_mpool_merge(&(results->mpool), &(results->record_mpool)) < 0)
    {
        neo4j_log_trace_errno(results->logger, "neo4j_mpool_merge failed");
        set_failure(results, errno);
        return -1;
    }

    return stream_end(results, type, "PULL_ALL", argv, argc);
}


int discard_all_callback(void *cdata, neo4j_message_type_t type,
        const neo4j_value_t *argv, uint16_t argc)
{
    assert(cdata != NULL);
    assert(argc == 0 || argv != NULL);
    run_result_stream_t *results = (run_result_stream_t *)cdata;

    --(results->refcount);
    results->streaming = false;

    return stream_end(results, type, "DISCARD_ALL", argv, argc);
}


int stream_end(run_result_stream_t *results, neo4j_message_type_t type,
        const char *src_message_type, const neo4j_value_t *argv, uint16_t argc)
{
    neo4j_logger_t *logger = results->logger;
    neo4j_connection_t *connection = results->connection;

    if (connection == NULL || type == NULL)
    {
        return 0;
    }

    neo4j_config_t *config = connection->config;

#ifndef NEOCLIENT_BUILD
    if (type == NEO4J_IGNORED_MESSAGE)
#else
    if ( MESSAGE_TYPE_IS(type,IGNORED) )
#endif
    {
        if (results->failure == 0)
        {
            neo4j_log_error(logger,
                    "Unexpected IGNORED message received in %p"
                    " (in response to %s, yet no failure occurred)",
                    (void *)connection, src_message_type);
            set_failure(results, errno = EPROTO);
            return -1;
        }
        return 0;
    }

    assert(results->failure == 0);

#ifndef NEOCLIENT_BUILD
    if (type == NEO4J_FAILURE_MESSAGE)
#else
    if ( MESSAGE_TYPE_IS(type,FAILURE) )
#endif
    {
        return set_eval_failure(results, src_message_type, argv, argc);
    }

#ifndef NEOCLIENT_BUILD
    if (type != NEO4J_SUCCESS_MESSAGE)
#else
    if ( !MESSAGE_TYPE_IS(type,SUCCESS) )
#endif
    {
        neo4j_log_error(logger,
                "Unexpected %s message received in %p"
                " (in response to %s)", neo4j_message_type_str(type),
                (void *)connection, src_message_type);