Neo4j-Client

 view release on metacpan or  search on metacpan

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

 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#include "../../config.h"
#include "connection.h"
#include "messages.h"
#include "buffering_iostream.h"
#include "chunking_iostream.h"
#include "client_config.h"
#include "deserialization.h"
#include "memory.h"
#include "metadata.h"
#include "network.h"
#ifdef HAVE_OPENSSL
#include "openssl_iostream.h"
#endif
#include "posix_iostream.h"
#include "serialization.h"
#include "transaction.h"
#include "util.h"
#include <assert.h>
#include <unistd.h>

#ifndef __has_attribute
#  define __has_attribute(x) 0  /* for GCC 4.8 and earlier */
#endif

static int add_userinfo_to_config(const char *userinfo, neo4j_config_t *config,
        uint_fast32_t flags);
static neo4j_connection_t *establish_connection(const char *hostname,
        unsigned int port, neo4j_config_t *config, uint_fast32_t flags);
static neo4j_iostream_t *std_tcp_connect(
        struct neo4j_connection_factory *factory, const char *hostname,
        unsigned int port, neo4j_config_t *config, uint_fast32_t flags,
        struct neo4j_logger *logger);
static int negotiate_protocol_version(neo4j_iostream_t *iostream,
				      version_spec_t *supported_versions,
        uint32_t *protocol_version, uint32_t *protocol_minor_version);

static bool interrupted(neo4j_connection_t *connection);
static int session_reset(neo4j_connection_t *connection);

static int send_requests(neo4j_connection_t *connection);
static int receive_responses(neo4j_connection_t *connection,
        const unsigned int *condition, bool interruptable);
static int drain_queued_requests(neo4j_connection_t *connection);

static struct neo4j_request *new_request(neo4j_connection_t *connection);
static void pop_request(neo4j_connection_t* connection);

static int initialize(neo4j_connection_t *connection);
static int initialize_callback(void *cdata, neo4j_message_type_t type,
        const neo4j_value_t *argv, uint16_t argc);
static int ack_failure(neo4j_connection_t *connection  );
static int ack_failure_callback(void *cdata, neo4j_message_type_t type,
       const neo4j_value_t *argv, uint16_t argc);

#if __has_attribute(unused)
static int hello(neo4j_connection_t *connection) __attribute__((unused));
#endif
static int goodbye(neo4j_connection_t *connection);

static neo4j_map_entry_t xtra[2];

struct neo4j_connection_factory neo4j_std_connection_factory =
{
    .tcp_connect = &std_tcp_connect
};


neo4j_connection_t *neo4j_connect(const char *uri_string,
        neo4j_config_t *config, uint_fast32_t flags)
{
    REQUIRE(uri_string != NULL, NULL);
    config = neo4j_config_dup(config);
    if (config == NULL)
    {
        return NULL;
    }

    neo4j_connection_t *connection = NULL;

    struct uri *uri = parse_uri(uri_string, NULL);
    if (uri == NULL)
    {
        if (errno == EINVAL)
        {
            errno = NEO4J_INVALID_URI;
        }
        goto failure;
    }

    if (uri->scheme == NULL ||
            (strcmp(uri->scheme, "neo4j") != 0 &&
             strcmp(uri->scheme, "bolt") != 0))
    {
        errno = NEO4J_UNKNOWN_URI_SCHEME;
        goto failure;
    }

    if (uri->userinfo != NULL)
    {
        if (!(flags & NEO4J_NO_URI_CREDENTIALS) &&
                add_userinfo_to_config(uri->userinfo, config, flags))
        {
            goto failure;
        }
        // clear any password in the URI
        size_t userinfolen = strlen(uri->userinfo);
        memset_s(uri->userinfo, userinfolen, 0, userinfolen);
    }

    unsigned int port = (uri->port > 0)? uri->port : NEO4J_DEFAULT_TCP_PORT;
    connection = establish_connection(uri->hostname, port, config, flags);
    if (connection == NULL)

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

 *         receiving of results.
 * @return 0 on success, -1 if a error occurs (errno will be set), 1 if
 *         interrupted, and >1 if a valid FAILURE message is received (in which
 *         case, all inflight requests will have been drained).
 */
int receive_responses(neo4j_connection_t *connection, const unsigned int *condition,
        bool interruptable)
{
    assert(connection != NULL);
    ENSURE_NOT_NULL(unsigned int, condition, 1);

    bool failure = false;
    while ((failure || *condition > 0) && connection->inflight_requests > 0 &&
            (!interruptable || !interrupted(connection)))
    {
        neo4j_message_type_t type;
        const neo4j_value_t *argv;
        uint16_t argc;

        struct neo4j_request *request =
            &(connection->request_queue[connection->request_queue_head]);
        if (neo4j_connection_recv(connection, request->mpool,
                    &type, &argv, &argc))
        {
            neo4j_log_trace_errno(connection->logger,
                    "neo4j_connection_recv failed");
            return -1;
        }

#ifndef NEOCLIENT_BUILD
        if (failure && type != NEO4J_IGNORED_MESSAGE)
#else
        if (failure && !MESSAGE_TYPE_IS(type,IGNORED))
#endif
        {
            neo4j_log_error(connection->logger,
                    "Unexpected %s message received in %p"
                    " (expected IGNORED after failure occurred)",
                    neo4j_message_type_str(type), (void *)connection);
            errno = EPROTO;
            connection->failed = true;
	    neo4j_log_trace(connection->logger,
			    "Connection failed with unexpected message after failure (receive_responses) on %p",
			    (void *)connection);
	    
            return -1;
        }
#ifndef NEOCLIENT_BUILD	
        if (type == NEO4J_FAILURE_MESSAGE)
#else
	if ( MESSAGE_TYPE_IS(type,FAILURE) )
#endif
        {
            failure = true;
        }

        neo4j_log_debug(connection->logger, "rcvd %s in response to %s (%p)",
                neo4j_message_type_str(type),
                neo4j_message_type_str(request->type), (void *)request);
        // request callback executed here
        int result = request->receive(request->cdata, type, argv, argc);
        int errsv = errno;
        if (result <= 0)
        {
            pop_request(connection);
            (connection->inflight_requests)--;
        }
        if (result < 0)
        {
	    char code[128];
	    neo4j_string_value(neo4j_map_get(argv[0],"code"),code,sizeof(code));
	    char msg[256];
	    neo4j_string_value(neo4j_map_get(argv[0],"message"),msg,sizeof(msg));
	    connection->failed = true;
	    neo4j_log_trace(connection->logger,
			    "Connection failed in request->receive() (receive_responses) on %p\nFailure code was: %s\nMessage was: %s",
			    (void *)connection, (const char *) code,
			    (const char *)msg);
	    errno = errsv;
	    return -1;
	}
    }

    if (interruptable && interrupted(connection))
    {
        return 1;
    }

    assert(!failure || connection->inflight_requests == 0);
    return failure? 2 : 0;
}


/**
 * Send IGNORED to all queued requests.
 *
 * @internal
 *
 * This will also generate IGNORED for all inflight requests, so this
 * method should only be called when there are no inflight requests or
 * when a terminal error has occured and the connection will be closed.
 *
 * @param [connection] The connection.
 * @return 0 on success, -1 if an error occurs (errno will be set).
 */
int drain_queued_requests(neo4j_connection_t *connection)
{
    assert(connection != NULL);

    int err = 0;
    int errsv = errno;
    while (connection->request_queue_depth > 0)
    {
        struct neo4j_request *request =
            &(connection->request_queue[connection->request_queue_head]);

        neo4j_log_trace(connection->logger, "draining %s (%p) from queue on %p",
                neo4j_message_type_str(request->type),
                (void *)request, (void *)connection);
        int result = request->receive(request->cdata, NULL, NULL, 0);
        assert(result <= 0);
        if (err == 0 && result < 0)
        {
            err = -1;
            errsv = errno;
        }
        pop_request(connection);
    }

    connection->inflight_requests = 0;
    errno = errsv;
    return err;
}


/**
 * Add a queued request.
 *
 * @internal
 *
 * The returned pointer will be to a request struct already added to the tail
 * of the queue. It MUST be populated with valid attributes before any other
 * connection methods are invoked.
 *
 * @param [connection] The connection.
 * @return The queued request, which MUST be populated with valid attributes,
 *         or `NULL` if an error occurs (errno will be set).
 */
struct neo4j_request *new_request(neo4j_connection_t *connection)
{
    assert(connection != NULL);
    const neo4j_config_t *config = connection->config;

    if (connection->failed)
    {
        errno = NEO4J_SESSION_FAILED;
        return NULL;
    }

    if (connection->request_queue_depth >= connection->request_queue_size)
    {
        assert(connection->request_queue_depth == connection->request_queue_size);
        errno = ENOBUFS;
        return NULL;
    }

    unsigned int request_queue_free =
        connection->request_queue_size - connection->request_queue_depth;
    unsigned int request_queue_tail =
        (request_queue_free > connection->request_queue_head)?
        connection->request_queue_head + connection->request_queue_depth :
        connection->request_queue_head - request_queue_free;

    (connection->request_queue_depth)++;
    struct neo4j_request *req =
        &(connection->request_queue[request_queue_tail]);
    req->_mpool = neo4j_mpool(config->allocator, config->mpool_block_size);
    req->mpool = &(req->_mpool);
    return req;
}


/**
 * Pop a request off the head of the queue.
 *
 * @internal
 *
 * @param [connection] The connection.
 */
void pop_request(neo4j_connection_t* connection)
{
    assert(connection != NULL);
    assert(connection->request_queue_depth > 0);

    struct neo4j_request *req =
        &(connection->request_queue[connection->request_queue_head]);
    neo4j_mpool_drain(&(req->_mpool));
    memset(req, 0, sizeof(struct neo4j_request));

    (connection->request_queue_depth)--;
    (connection->request_queue_head)++;
    if (connection->request_queue_head >= connection->request_queue_size)
    {
        assert(connection->request_queue_head == connection->request_queue_size);
        connection->request_queue_head = 0;
    }
}


struct init_cdata
{
    neo4j_connection_t *connection;
    int error;
};

int initialize(neo4j_connection_t *connection)
{
    assert(connection != NULL);
    neo4j_config_t *config = connection->config;

    char host[NEO4J_MAXHOSTLEN];
    if (describe_host(host, sizeof(host), connection->hostname,
            connection->port))
    {
        return -1;
    }

    if (ensure_basic_auth_credentials(config, host))
    {
        return -1;
    }

    int err = -1;

    struct neo4j_request *req = new_request(connection);
    if (req == NULL)
    {
        goto cleanup;
    }

    struct init_cdata cdata = { .connection = connection, .error = 0 };

    req->type = NEO4J_INIT_MESSAGE;
    if (connection->version < 3)
      {
      req->_argv[0] = neo4j_string(config->client_id);
      neo4j_map_entry_t auth_token[3] =
        { neo4j_map_entry("scheme", neo4j_string("basic")),
          neo4j_map_entry("principal", neo4j_string(config->username)),
          neo4j_map_entry("credentials", neo4j_string(config->password)) };
      req->_argv[1] = neo4j_map(auth_token, 3);
      req->argv = req->_argv;
      req->argc = 2;
      }
    else if (connection->version == 4 && connection->minor_version == 4)
      {
        neo4j_value_t patch_bolt[1] = {neo4j_string("utc")};
        neo4j_map_entry_t auth_token[5] =
          { neo4j_map_entry("user_agent", neo4j_string(config->client_id)),
            neo4j_map_entry("scheme", neo4j_string("basic")),
            neo4j_map_entry("principal", neo4j_string(config->username)),
            neo4j_map_entry("credentials", neo4j_string(config->password)),
            neo4j_map_entry("patch_bolt", neo4j_list(patch_bolt, 1)) };
        req->_argv[0] = neo4j_map(auth_token, 5);
        req->argv = req->_argv;
        req->argc = 1;
      }
    else
      {
        neo4j_map_entry_t auth_token[4] =
          { neo4j_map_entry("user_agent", neo4j_string(config->client_id)),
            neo4j_map_entry("scheme", neo4j_string("basic")),
            neo4j_map_entry("principal", neo4j_string(config->username)),
            neo4j_map_entry("credentials", neo4j_string(config->password)) };
        req->_argv[0] = neo4j_map(auth_token, 4);
        req->argv = req->_argv;
        req->argc = 1;
      }
    req->receive = initialize_callback;
    req->cdata = &cdata;

    neo4j_log_trace(connection->logger,
                    (connection->version < 3)?
            "enqu INIT{\"%s\", {scheme: basic, principal: \"%s\", "
                    "credentials: ****}} (%p) in %p" :
            "enqu INIT{user_agent: \"%s\", scheme: basic, principal: \"%s\", "
                    "credentials: ****}} (%p) in %p",
            config->client_id, config->username,
            (void *)req, (void *)connection);

    if (neo4j_session_sync(connection, NULL))
    {
        if (cdata.error != 0)
        {
            errno = cdata.error;
        }
        goto cleanup;
    }

    if (cdata.error != 0)
    {
        assert(cdata.error == NEO4J_INVALID_CREDENTIALS ||
                cdata.error == NEO4J_AUTH_RATE_LIMIT);
        errno = cdata.error;
        goto cleanup;
    }

    err = 0;

    int errsv;
cleanup:
    errsv = errno;
    // clear password out of connection config
    ignore_unused_result(neo4j_config_set_password(connection->config, NULL));
    errno = errsv;
    return err;
}

// hello - alias for initialize (Bolt 3.0)

#if __has_attribute(unused)
int hello(neo4j_connection_t *connection)
{
    return initialize(connection);
}
#endif

// Note (Bolt 3.0)
// The GOODBYE message does not generate a server response. It signals a graceful
// finish on the client-side. This functon is called in neo4j_close() when
// Bolt 3+ is used

int goodbye(neo4j_connection_t *connection)
{
    REQUIRE(connection != NULL, -1);
    if (connection->failed)
      {
        errno = NEO4J_SESSION_FAILED;
        return -1;
      }

    if (neo4j_connection_send(connection, NEO4J_GOODBYE_MESSAGE, NULL, 0))
      {
        connection->failed = true;
	neo4j_log_trace(connection->logger,
			"Connection failed in neo4j_connection_send() (goodbye) on %p",
			(void *)connection);
	
        return -1;
      }
    neo4j_log_trace(connection->logger, "sent GOODBYE in %p", (void *)connection);
    return 0;
}

int initialize_callback(void *cdata, neo4j_message_type_t type,
        const neo4j_value_t *argv, uint16_t argc)
{
    if (type == NULL)
    {
        return 0;
    }

    assert(cdata != NULL);
    neo4j_connection_t *connection = ((struct init_cdata *)cdata)->connection;

    char description[128];

#ifndef NEOCLIENT_BUILD
    if (type == NEO4J_SUCCESS_MESSAGE)
#else
    if ( MESSAGE_TYPE_IS(type,SUCCESS) )
#endif
    {
        snprintf(description, sizeof(description),
                "SUCCESS in %p (response to INIT)", (void *)connection);
        const neo4j_value_t *metadata = neo4j_validate_metadata(argv, argc,
                description, connection->logger);
        if (metadata == NULL)
        {
            return -1;
        }
        if (neo4j_log_is_enabled(connection->logger, NEO4J_LOG_TRACE))
        {
            neo4j_metadata_log(connection->logger, NEO4J_LOG_TRACE, description,
                    *metadata);
        }
        neo4j_value_t ce = neo4j_map_get(*metadata, "credentials_expired");
        connection->credentials_expired =
                (neo4j_type(ce) == NEO4J_BOOL && neo4j_bool_value(ce));
        neo4j_value_t si = neo4j_map_get(*metadata, "server");
        if (neo4j_type(si) == NEO4J_STRING)
        {
            connection->server_id = strndup(neo4j_ustring_value(si),
                    neo4j_string_length(si));
            if (connection->server_id == NULL)
            {
                return -1;
            }
        }
        return 0;
    }

#ifndef NEOCLIENT_BUILD
    if (type != NEO4J_FAILURE_MESSAGE)
#else
    if ( !MESSAGE_TYPE_IS(type,FAILURE) )
#endif
    {
        neo4j_log_error(connection->logger,
                "Unexpected %s message received in %p"
                " (expected SUCCESS in response to INIT)",
                neo4j_message_type_str(type), (void *)connection);
        errno = EPROTO;
        return -1;
    }

    // handle failure
    snprintf(description, sizeof(description),
            "FAILURE in %p (response to INIT)", (void *)connection);
    const neo4j_value_t *metadata = neo4j_validate_metadata(argv, argc,
            description, connection->logger);
    if (metadata == NULL)
    {
        return -1;
    }

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

    const neo4j_config_t *config = connection->config;
    struct neo4j_failure_details details;
    neo4j_mpool_t mpool =
        neo4j_mpool(config->allocator, config->mpool_block_size);
    if (neo4j_meta_failure_details(&details, *metadata, &mpool,
                description, connection->logger))
    {
        return -1;
    }

    int result = -1;

    if (strcmp("Neo.ClientError.Security.EncryptionRequired",
            details.code) == 0)
    {
        errno = NEO4J_SERVER_REQUIRES_SECURE_CONNECTION;
        goto cleanup;
    }
    if (strcmp("Neo.ClientError.Security.Unauthorized", details.code) == 0)
    {
        ((struct init_cdata *)cdata)->error = NEO4J_INVALID_CREDENTIALS;
        result = 0;
        goto cleanup;
    }
    if (strcmp("Neo.ClientError.Security.AuthenticationRateLimit",
            details.code) == 0)
    {
        ((struct init_cdata *)cdata)->error = NEO4J_AUTH_RATE_LIMIT;
        result = 0;
        goto cleanup;
    }

    neo4j_log_error(connection->logger, "Session initialization failed: %s",
            details.message);
    errno = NEO4J_UNEXPECTED_ERROR;

cleanup:
    neo4j_mpool_drain(&mpool);
    return result;
}

// in Bolt 3, ACK_FAILURE is eschewed for RESET message as
// a response to server FAILURE.
int ack_failure(neo4j_connection_t *connection)
{
    assert(connection != NULL);

    struct neo4j_request *req = new_request(connection);
    if (req == NULL)
    {
        return -1;
    }
    req->type = (connection->version < 3)? NEO4J_ACK_FAILURE_MESSAGE :
      NEO4J_RESET_MESSAGE;
    req->argc = 0;
    req->receive = ack_failure_callback;
    req->cdata = connection;

    neo4j_log_trace(connection->logger, "enqu %s (%p) in %p",
            (connection->version < 3)? "ACK_FAILURE" : "RESET",
            (void *)req, (void *)connection);

    return neo4j_session_sync(connection, NULL);
}


int ack_failure_callback(void *cdata, neo4j_message_type_t type,
        const neo4j_value_t *argv, uint16_t argc)
{
    assert(cdata != NULL);
    neo4j_connection_t *connection = (neo4j_connection_t *)cdata;

    char buf[12];
    strcpy(buf, (connection->version < 3)? "ACK_FAILURE" : "RESET");
#ifndef NEOCLIENT_BUILD
    if (type == NEO4J_IGNORED_MESSAGE || type == NULL)
#else
    if ( MESSAGE_TYPE_IS(type,IGNORED) || type == NULL )
#endif
    {
        // only when draining after connection close
        return 0;
    }
#ifndef NEOCLIENT_BUILD    
    if (type != NEO4J_SUCCESS_MESSAGE)
#else
    if ( !MESSAGE_TYPE_IS(type,SUCCESS) )
#endif
    {
        neo4j_log_error(connection->logger,
                "Unexpected %s message received in %p"
                " (expected SUCCESS in response to %s)",
                 neo4j_message_type_str(type), (void *)connection, buf);
        errno = EPROTO;
        return -1;
    }

    neo4j_log_trace(connection->logger, "%s complete in %p", buf,
            (void *)connection);
    return 0;
}


int neo4j_session_run(neo4j_connection_t *connection, neo4j_mpool_t *mpool,
        const char *statement, neo4j_value_t params, neo4j_value_t extra,
        neo4j_response_recv_t callback, void *cdata)
{
    REQUIRE(connection != NULL, -1);
    REQUIRE(mpool != NULL, -1);
    REQUIRE(statement != NULL, -1);
    REQUIRE(neo4j_type(params) == NEO4J_MAP || neo4j_is_null(params), -1);
    REQUIRE(neo4j_type(extra) == NEO4J_MAP || neo4j_is_null(extra), -1);
    REQUIRE(callback != NULL, -1);

    if (neo4j_atomic_bool_set(&(connection->processing), true))
    {
        errno = NEO4J_SESSION_BUSY;
        return -1;
    }

    int err = -1;
    struct neo4j_request *req = new_request(connection);
    if (req == NULL)
    {
        goto cleanup;
    }
    req->type = NEO4J_RUN_MESSAGE;
    req->_argv[0] = neo4j_string(statement);
    req->_argv[1] = neo4j_is_null(params)? neo4j_map(NULL, 0) : params;
    req->_argv[2] = neo4j_is_null(extra)? neo4j_map(NULL,0) : extra;
    req->argv = req->_argv;
    req->argc = (connection->version > 2)? 3 : 2;
    req->mpool = mpool;
    req->receive = callback;
    req->cdata = cdata;

    if (neo4j_log_is_enabled(connection->logger, NEO4J_LOG_TRACE))
    {
        char buf[1024];
        if (connection->version < 3)
          {
            neo4j_log_trace(connection->logger, "enqu RUN{\"%s\", %s} (%p) in %p",
                            statement, neo4j_tostring(req->argv[1], buf, sizeof(buf)),
                            (void *)req, (void *)connection);
          }
        else
          {
            char buf2[1024];
            neo4j_log_trace(connection->logger, "enqu RUN{\"%s\", %s, %s} (%p) in %p",
                            statement, neo4j_tostring(req->argv[1], buf, sizeof(buf)),
                            neo4j_tostring(req->argv[2], buf2, sizeof(buf2)),
                            (void *)req, (void *)connection);
          }
    }

    err = 0;

cleanup:
    neo4j_atomic_bool_set(&(connection->processing), false);
    return err;
}

int neo4j_session_pull_all(neo4j_connection_t *connection, int n, int qid, 
        neo4j_mpool_t *mpool, neo4j_response_recv_t callback, void *cdata)
{
    REQUIRE(connection != NULL, -1);
    REQUIRE(mpool != NULL, -1);
    REQUIRE(callback != NULL, -1);

    if (neo4j_atomic_bool_set(&(connection->processing), true))
    {
        errno = NEO4J_SESSION_BUSY;
        return -1;
    }

    int err = -1;
    struct neo4j_request *req = new_request(connection);
    if (req == NULL)
    {
        goto cleanup;
    }

    req->type = NEO4J_PULL_ALL_MESSAGE;
    if (connection->version < 4)
      {
        req->argv = NULL;
        req->argc = 0;
      }
    else
      {
        xtra[0] = neo4j_map_entry("n",neo4j_int( n));
        xtra[1] = neo4j_map_entry("qid",neo4j_int( qid));
        req->_argv[0] = neo4j_map(xtra, 2);
        req->argv = req->_argv;
        req->argc = 1;
      }
    req->mpool = mpool;
    req->receive = callback;
    req->cdata = cdata;
    if (connection->version < 4)
      {
        neo4j_log_trace(connection->logger, "enqu PULL_ALL (%p) in %p",
		        (void *)req, (void *)connection);
      }
    else
      {
        char buf[128];
        neo4j_log_trace(connection->logger, "enqu PULL %s (%p) in %p",
                neo4j_tostring(req->argv[0],buf, sizeof(buf)), (void *)req, (void *)connection);
      }

    err = 0;

cleanup:
    neo4j_atomic_bool_set(&(connection->processing), false);
    return err;
}


int neo4j_session_discard_all(neo4j_connection_t *connection, int n, int qid,
        neo4j_mpool_t *mpool, neo4j_response_recv_t callback, void *cdata)
{
    REQUIRE(connection != NULL, -1);
    REQUIRE(mpool != NULL, -1);
    REQUIRE(callback != NULL, -1);

    if (neo4j_atomic_bool_set(&(connection->processing), true))
    {
        errno = NEO4J_SESSION_BUSY;
        return -1;
    }

    int err = -1;
    struct neo4j_request *req = new_request(connection);
    if (req == NULL)
    {
        goto cleanup;
    }

    req->type = NEO4J_DISCARD_ALL_MESSAGE;
    if (connection->version < 4)
      {
        req->argv = NULL;
        req->argc = 0;
      }
    else
      {
        xtra[0] = neo4j_map_entry("n",neo4j_int(n));
        xtra[1] = neo4j_map_entry("qid",neo4j_int(qid));
        req->_argv[0] = neo4j_map(xtra, 2);
        req->argv = req->_argv;
        req->argc = 1;
      }
    req->mpool = mpool;
    req->receive = callback;
    req->cdata = cdata;

    if (connection->version < 4)
      {
        neo4j_log_trace(connection->logger, "enqu DISCARD_ALL (%p) in %p",
                        (void *)req, (void *)connection);
      }
    else
      {
        char buf[128];
        neo4j_log_trace(connection->logger, "enqu DISCARD %s (%p) in %p",
                        neo4j_tostring(req->argv[0],buf, sizeof(buf)), (void *)req, (void *)connection);
      }

    err = 0;

cleanup:
    neo4j_atomic_bool_set(&(connection->processing), false);
    return err;
}

int neo4j_session_transact(neo4j_connection_t *connection, const char*msg_type, neo4j_response_recv_t callback, void *cdata)
{
    REQUIRE(connection != NULL, -1);
    REQUIRE(cdata != NULL, -1);
    REQUIRE(callback != NULL, -1);

    neo4j_transaction_t *tx = (neo4j_transaction_t *) cdata;
    int err = -1;
    struct neo4j_request *req = new_request(connection);
    if (req == NULL)
    {
        goto cleanup;
    }

    req->type = neo4j_message_type_for_type(msg_type);
    if (strcmp(msg_type,"BEGIN") == 0)
    {
	const neo4j_map_entry_t ent[3] = {
	    neo4j_map_entry("mode",neo4j_string(tx->mode)),
	    neo4j_map_entry("db",neo4j_string(tx->dbname == NULL ? "" : tx->dbname)),
	    neo4j_map_entry("tx_timeout",neo4j_int(tx->timeout))
	};
	int nent = (tx->timeout >= 0)?3:2;
	req->_argv[0] = neo4j_map(ent,nent); // extra dictionary
	req->argv = req->_argv;
	req->argc = 1;
    }
    else
    {
	req->argv = NULL;
	req->argc = 0;
    }
    req->mpool = &(tx->mpool);
    req->receive = callback; // callback specified in transaction.c
    req->cdata = cdata; // this will be the neo4j_transaction_t object
    if (neo4j_log_is_enabled(connection->logger,NEO4J_LOG_TRACE)) {
        neo4j_log_trace(connection->logger, "enqu %s (%p) in %p",
                        msg_type, (void *)req, (void *)connection);
    }
    if (neo4j_session_sync(connection,NULL)) {
      if (tx->failed != 0) {
        errno = tx->failure;
      }
      goto cleanup;
    }

    err = 0;

cleanup:
    return err;
}

int parse_version_string(char *version_string, version_spec_t *vs) {
  int M0=0, m0=0, M1=0, m1=0,  n;
  vs->major=0;
  vs->minor=0;
  vs->and_lower=0;
  n = sscanf(version_string, "%d.%d-%d.%d", &M0, &m0, &M1, &m1);
  if (n==1) {
    if (sscanf(version_string, "%d-%d.%d", &M0, &M1, &m1)==3) {
      m0 = 0;
      n = 4;
    }
  }
  switch (n) {
  case 1:
    vs->major = M0;
    vs->minor = 0;
    break;
  case 2:
    vs->major = M0;
    vs->minor = m0;    
    break;
  case 3:
    m1 = 0;
#if defined(__STDC_VERSION__) && __STDC_VERSION__ >= 202311L
    [[fallthrough]];
#elif __has_attribute(fallthrough)
    __attribute__((fallthrough));
#endif
    /* FALLTHROUGH */
  case 4:
    if (M0 != M1) {
      // Only minor version ranges within a single major version allowed
      return -1;
    }
    vs->major = M0;
    vs->minor = m0;    
    if (m1 >= m0) {
      vs->minor = m1;
      vs->and_lower = m1-m0;
    }
    else {
      vs->and_lower = m0-m1;
    }



( run in 0.495 second using v1.01-cache-2.11-cpan-7fcb06a456a )