view release on metacpan or  search on metacpan

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

/* vi:set ts=4 sw=4 expandtab:
 *
 * Copyright 2016, Chris Leishman (http://github.com/cleishm)
 *
 * 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 "transaction.h"
#include "connection.h"
#include "neo4j-client.h"
#include "memory.h"
#include "result_stream.h"
#include "util.h"
#include "values.h"
#include "atomic.h"
#include <assert.h>
#include <string.h>
#include <stddef.h>
#include <unistd.h>
#include <stdio.h>

neo4j_transaction_t *new_transaction(neo4j_config_t *config, neo4j_connection_t *connection, int timeout, const char *mode, const char *dbname);
int begin_callback(void *cdata, neo4j_message_type_t type, const neo4j_value_t *argv, uint16_t argc);
int commit_callback(void *cdata, neo4j_message_type_t type, const neo4j_value_t *argv, uint16_t argc);
int rollback_callback(void *cdata, neo4j_message_type_t type, const neo4j_value_t *argv, uint16_t argc);
int tx_failure(neo4j_transaction_t *tx);
bool tx_defunct(neo4j_transaction_t *tx);
int tx_commit(neo4j_transaction_t *tx);
int tx_rollback(neo4j_transaction_t *tx);
neo4j_result_stream_t *tx_run(neo4j_transaction_t *tx, const char *statement, neo4j_value_t params, int send);


// rough out the transaction based calls
// there will be bookkeeping to do - handling errors when server state
// is mismatch with the request:
// - when server is READY but run_in_tx is exec'd
// - when server is STREAMING but run is exec'd <- may be handled by the results->starting,
//   results->streaming flags
// - when server is TX_READY but run or send is exec'd
// - when server is TX_STREAMING but run or send is exec'd
// erroring when negotiated protocol is not 3+

// Note, BEGIN, COMMIT, ROLLBACK responses don't belong on a results stream. So maybe
// should have a transaction structure, analogous to the results structure, that
// stores info like success responses, failure responses, bookmarks.

// begin_tx - specify timeout and mode, but ignore bookmarks and metadata ATM
// must check neo4j_tx_failure(tx)
neo4j_transaction_t *neo4j_begin_tx(neo4j_connection_t *connection,
        int tx_timeout, const char *tx_mode, const char *dbname)
{
    REQUIRE(connection != NULL, NULL);

    neo4j_config_t *config = connection->config;
    if (connection->version < 3)
      {
        errno = NEO4J_FEATURE_UNAVAILABLE;
        char ebuf[256];
        neo4j_log_error(connection->logger,
                "Cannot create transaction on %p: %s\n", (void *)connection,
                        neo4j_strerror(errno, ebuf, sizeof(ebuf)));
        return NULL;
      }
    neo4j_transaction_t *tx = new_transaction(config, connection, tx_timeout, tx_mode, dbname);
    if (neo4j_session_transact(connection, "BEGIN", begin_callback, tx))
      {
        neo4j_log_error_errno(tx->logger, "tx begin failed");
        tx->failed = 1;
        tx->failure = errno;
      }
    return tx;
}

int begin_callback(void *cdata, neo4j_message_type_t type, const neo4j_value_t *argv, uint16_t argc)
{
  assert(cdata != NULL);
  assert(argc == 0 || argv != NULL);
  neo4j_transaction_t *tx = (neo4j_transaction_t *) cdata;

#ifndef NEOCLIENT_BUILD  
  if (type == NEO4J_FAILURE_MESSAGE)
#else
  if ( MESSAGE_TYPE_IS(type,FAILURE) )
#endif
    {
      // get FAILURE argv and set tx failure info here
      tx->failed = 1;
      tx->failure = NEO4J_TRANSACTION_FAILED;
      tx->failure_code = neo4j_map_get(argv[0],"code");
      tx->failure_message = neo4j_map_get(argv[0],"message");
      errno = tx->failure;
      neo4j_log_error_errno(tx->logger, "tx begin failed");

      return -1;
    }

#ifndef NEOCLIENT_BUILD  
  if (type == NEO4J_IGNORED_MESSAGE)
#else
  if ( MESSAGE_TYPE_IS(type,IGNORED) )
#endif
    {
      neo4j_log_trace(tx->logger, "tx begin ignored");
      return 0;
    }
  char description[128];
  snprintf(description, sizeof(description), "%s in %p (response to BEGIN)",
           neo4j_message_type_str(type), (void *)tx->connection);

#ifndef NEOCLIENT_BUILD  
  if (type != NEO4J_SUCCESS_MESSAGE)
#else
  if ( !MESSAGE_TYPE_IS(type,SUCCESS) )
#endif
    {
      neo4j_log_error(tx->logger, "Unexpected %s", description);
      tx->failed = 1;
      tx->failure = EPROTO;
      errno = tx->failure;
      return -1;
    }
  tx->is_open = 1;
  return 0;
}

// commit_tx
int tx_commit(neo4j_transaction_t *tx)
{
    REQUIRE(tx != NULL, -1);
    if (neo4j_session_transact(tx->connection, "COMMIT", commit_callback, tx))
      {
        neo4j_log_error_errno(tx->logger, "tx commit failed");
        tx->failed = 1;
        tx->failure = errno;
      }
    else
    {
	tx->is_open = 0;
	tx->failed = 0;
	tx->failure_code = neo4j_null;
	tx->failure_message = neo4j_null;	
    }
    return -tx->failed;
}

int commit_callback(void *cdata, neo4j_message_type_t type, const neo4j_value_t *argv, uint16_t argc)
{
  assert(cdata != NULL);
  assert(argc == 0 || argv != NULL);
  neo4j_transaction_t *tx = (neo4j_transaction_t *) cdata;

#ifndef NEOCLIENT_BUILD  
  if (type == NEO4J_FAILURE_MESSAGE)
#else
  if ( MESSAGE_TYPE_IS(type,FAILURE) )
#endif
    {
      // get FAILURE argv and set tx failure info here
      tx->failed = 1;
      tx->failure = NEO4J_TRANSACTION_FAILED;
      tx->failure_code = neo4j_map_get(argv[0],"code");
      tx->failure_message = neo4j_map_get(argv[0],"message");
      // check here if transaction timed out; if so, set tx->is_expired
      errno = tx->failure;
      neo4j_log_error_errno(tx->logger, "tx commit failed");
      return -1;
    }
#ifndef NEOCLIENT_BUILD  
  if (type == NEO4J_IGNORED_MESSAGE)
#else
  if ( MESSAGE_TYPE_IS(type,IGNORED) )
#endif
    {
      neo4j_log_trace(tx->logger, "tx commit ignored");
      return 0;
    }
  char description[128];
  snprintf(description, sizeof(description), "%s in %p (response to COMMIT)",
           neo4j_message_type_str(type), (void *)tx->connection);

#ifndef NEOCLIENT_BUILD  
  if (type != NEO4J_SUCCESS_MESSAGE)
#else
  if ( !MESSAGE_TYPE_IS(type,SUCCESS) )
#endif
    {
      neo4j_log_error(tx->logger, "Unexpected %s", description);
      tx->failed = -1;
      tx->failure = EPROTO;
      errno = tx->failure;
      return -1;
    }
  if (argc) {
    neo4j_value_t svr_extra = argv[0];
    neo4j_value_t bookmark = neo4j_map_get(svr_extra,"bookmark");
    if (!neo4j_is_null(bookmark))
      {
        tx->commit_bookmark = bookmark;
      }
  }
  tx->is_open = 0;
  return 0;
}

// rollback_tx
// must check tx->failed after call
int tx_rollback(neo4j_transaction_t *tx)
{
  REQUIRE(tx != NULL, -1);
  if (neo4j_session_transact(tx->connection, "ROLLBACK", rollback_callback, tx))
    {
      neo4j_log_error_errno(tx->logger, "tx rollback failed");
      tx->failed = 1;
      tx->failure = errno;
    }
  else {
      tx->is_open = 0;
      tx->failed = 0;
      tx->failure_code = neo4j_null;
      tx->failure_message = neo4j_null;
  }
  return -tx->failed;
}

int rollback_callback(void *cdata, neo4j_message_type_t type, const neo4j_value_t *argv, uint16_t argc)
{
  assert(cdata != NULL);
  assert(argc == 0 || argv != NULL);
  neo4j_transaction_t *tx = (neo4j_transaction_t *) cdata;
#ifndef NEOCLIENT_BUILD  
  if (type == NEO4J_FAILURE_MESSAGE)
#else
  if ( MESSAGE_TYPE_IS(type,FAILURE) )
#endif
    {
      // get FAILURE argv and set tx failure info here
      tx->failed = 1;
      tx->failure = NEO4J_TRANSACTION_FAILED;
      tx->failure_code = neo4j_map_get(argv[0],"code");
      tx->failure_message = neo4j_map_get(argv[0],"message");
      // check here if transaction timed out; if so, set tx->is_expired
      errno = tx->failure;
      neo4j_log_error_errno(tx->logger, "tx rollback failed");
      return -1;
    }
#ifndef NEOCLIENT_BUILD  
  if (type == NEO4J_IGNORED_MESSAGE)
#else
  if ( MESSAGE_TYPE_IS(type,IGNORED) )
#endif
    {
      neo4j_log_trace(tx->logger, "tx rollback ignored");
      return 0;
    }
  char description[128];
  snprintf(description, sizeof(description), "%s in %p (response to ROLLBACK)",
           neo4j_message_type_str(type), (void *)tx->connection);

#ifndef NEOCLIENT_BUILD  
  if (type != NEO4J_SUCCESS_MESSAGE)
#else
  if ( !MESSAGE_TYPE_IS(type,SUCCESS) )
#endif
    {
      neo4j_log_error(tx->logger, "Unexpected %s", description);
      tx->failed = 1;
      fprintf(stderr,"Unexpected %s", description);
      tx->failure = EPROTO;
      errno = tx->failure;
      return -1;
    }

  // Bolt 3.0 spec sez SUCCESS argv "may contain metadata relating to the outcome". Pfft.
  tx->is_open = 0;
  tx->failed = 0;
  return 0;
}

// run_in_tx
// returns a result stream, namely, tx->results
// returns NULL and expires tx if tx has timed out

neo4j_result_stream_t *tx_run(neo4j_transaction_t *tx,
       const char *statement, neo4j_value_t params, int send)
{
  REQUIRE(tx != NULL, NULL);

  if (!neo4j_tx_is_open(tx) || neo4j_tx_defunct(tx)) {
      errno = NEO4J_TRANSACTION_DEFUNCT;