Mirrors
/
monero
mirror of https://github.com/monero-project/monero.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Commit tx IPC

This commit is contained in:
Oran Juice 2015-03-31 21:50:38 +05:30
parent 322900b374
commit bcbc24a456
No known key found for this signature in database
GPG Key ID: 71C5AF46CCB28124
13 changed files with 1031 additions and 173 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

117
src/ipc/daemon_ipc_handlers.cpp
View File

@ -161,7 +161,49 @@ namespace IPC
void send_raw_transaction(wap_proto_t *message) void send_raw_transaction(wap_proto_t *message)
{ {
if (!check_core_busy()) {
wap_proto_set_status(message, STATUS_CORE_BUSY);
return;
}
std::string tx_blob;
zchunk_t *tx_as_hex_chunk = wap_proto_tx_as_hex(message);
char *tx_as_hex = (char*)zchunk_data(tx_as_hex_chunk);
std::string tx_as_hex_string(tx_as_hex, zchunk_size(tx_as_hex_chunk));
if (!string_tools::parse_hexstr_to_binbuff(tx_as_hex_string, tx_blob))
{
LOG_PRINT_L0("[on_send_raw_tx]: Failed to parse tx from hexbuff: " << tx_as_hex_string);
wap_proto_set_status(message, STATUS_INVALID_TX);
return;
}
cryptonote::cryptonote_connection_context fake_context = AUTO_VAL_INIT(fake_context);
cryptonote::tx_verification_context tvc = AUTO_VAL_INIT(tvc);
if (!core->handle_incoming_tx(tx_blob, tvc, false))
{
LOG_PRINT_L0("[on_send_raw_tx]: Failed to process tx");
wap_proto_set_status(message, STATUS_INVALID_TX);
return;
}
if (tvc.m_verifivation_failed)
{
LOG_PRINT_L0("[on_send_raw_tx]: tx verification failed");
wap_proto_set_status(message, STATUS_TX_VERIFICATION_FAILED);
return;
}
if (!tvc.m_should_be_relayed)
{
LOG_PRINT_L0("[on_send_raw_tx]: tx accepted, but not relayed");
wap_proto_set_status(message, STATUS_TX_NOT_RELAYED);
return;
}
cryptonote::NOTIFY_NEW_TRANSACTIONS::request r;
r.txs.push_back(tx_blob);
core->get_protocol()->relay_transactions(r, fake_context);
//TODO: make sure that tx has reached other nodes here, probably wait to receive reflections from other nodes
wap_proto_set_status(message, STATUS_OK);
} }
void get_output_indexes(wap_proto_t *message) void get_output_indexes(wap_proto_t *message)
@ -186,5 +228,80 @@ namespace IPC
wap_proto_set_o_indexes(message, &frame); wap_proto_set_o_indexes(message, &frame);
wap_proto_set_status(message, STATUS_OK); wap_proto_set_status(message, STATUS_OK);
} }
void get_random_outs(wap_proto_t *message) {
if (!check_core_busy()) {
wap_proto_set_status(message, STATUS_CORE_BUSY);
return;
}
// The core does its stuff with old style RPC objects.
// So we construct and read from those objects.
cryptonote::COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::request req;
uint64_t outs_count = wap_proto_outs_count(message);
req.outs_count = outs_count;
zframe_t *amounts_frame = wap_proto_amounts(message);
uint64_t amounts_count = zframe_size(amounts_frame) / sizeof(uint64_t);
uint64_t *amounts = (uint64_t*)zframe_data(amounts_frame);
for (unsigned int i = 0; i < amounts_count; i++) {
req.amounts.push_back(amounts[i]);
}
cryptonote::COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::response res;
if (!core->get_random_outs_for_amounts(req, res))
{
wap_proto_set_status(message, STATUS_RANDOM_OUTS_FAILED);
}
// We have to convert the result into a JSON string.
rapidjson::Document result_json;
result_json.SetObject();
rapidjson::Document::AllocatorType &allocator = result_json.GetAllocator();
rapidjson::Value outputs_json(rapidjson::kArrayType);
typedef cryptonote::COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::outs_for_amount outs_for_amount;
typedef cryptonote::COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::out_entry out_entry;
for (unsigned int i = 0; i < res.outs.size(); i++) {
rapidjson::Value output(rapidjson::kObjectType);
outs_for_amount out = res.outs[i];
rapidjson::Value output_entries(rapidjson::kArrayType);
for (std::list<out_entry>::iterator it = out.outs.begin(); it != out.outs.end(); it++) {
rapidjson::Value output_entry(rapidjson::kObjectType);
out_entry entry = *it;
output_entry.AddMember("global_amount_index", entry.global_amount_index, allocator);
rapidjson::Value string_value(rapidjson::kStringType);
string_value.SetString(entry.out_key.data, 32, allocator);
output_entry.AddMember("out_key", string_value.Move(), allocator);
output_entries.PushBack(output_entry, allocator);
}
output.AddMember("amount", out.amount, allocator);
output.AddMember("outs", output_entries, allocator);
outputs_json.PushBack(output, allocator);
}
result_json.AddMember("outputs", outputs_json, allocator);
rapidjson::StringBuffer buffer;
rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
result_json.Accept(writer);
std::string block_string = buffer.GetString();
std::cout << block_string << std::endl;
zframe_t *frame = zframe_new(block_string.c_str(), block_string.length());
wap_proto_set_random_outputs(message, &frame);
std::stringstream ss;
std::for_each(res.outs.begin(), res.outs.end(), [&](outs_for_amount& ofa)
{
ss << "[" << ofa.amount << "]:";
CHECK_AND_ASSERT_MES(ofa.outs.size(), ;, "internal error: ofa.outs.size() is empty for amount " << ofa.amount);
std::for_each(ofa.outs.begin(), ofa.outs.end(), [&](out_entry& oe)
{
ss << oe.global_amount_index << " ";
});
ss << ENDL;
});
std::string s = ss.str();
LOG_PRINT_L2("COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS: " << ENDL << s);
wap_proto_set_status(message, STATUS_OK);
}
} }
} }

5
src/ipc/include/daemon_ipc_handlers.h
View File

@ -60,12 +60,17 @@ namespace IPC
const uint64_t STATUS_MINING_NOT_STARTED = 3; const uint64_t STATUS_MINING_NOT_STARTED = 3;
const uint64_t STATUS_WRONG_BLOCK_ID_LENGTH = 4; const uint64_t STATUS_WRONG_BLOCK_ID_LENGTH = 4;
const uint64_t STATUS_INTERNAL_ERROR = 5; const uint64_t STATUS_INTERNAL_ERROR = 5;
const uint64_t STATUS_INVALID_TX = 6;
const uint64_t STATUS_TX_VERIFICATION_FAILED = 7;
const uint64_t STATUS_TX_NOT_RELAYED = 8;
const uint64_t STATUS_RANDOM_OUTS_FAILED = 9;
namespace Daemon namespace Daemon
{ {
void start_mining(wap_proto_t *message); void start_mining(wap_proto_t *message);
void retrieve_blocks(wap_proto_t *message); void retrieve_blocks(wap_proto_t *message);
void send_raw_transaction(wap_proto_t *message); void send_raw_transaction(wap_proto_t *message);
void get_output_indexes(wap_proto_t *message); void get_output_indexes(wap_proto_t *message);
void get_random_outs(wap_proto_t *message);
void init(cryptonote::core *p_core, void init(cryptonote::core *p_core,
nodetool::node_server<cryptonote::t_cryptonote_protocol_handler<cryptonote::core> > *p_p2p, nodetool::node_server<cryptonote::t_cryptonote_protocol_handler<cryptonote::core> > *p_p2p,
bool p_testnet); bool p_testnet);

17
src/ipc/include/wap_client.h
View File

@ -54,6 +54,12 @@ WAP_EXPORT zactor_t *
WAP_EXPORT zsock_t * WAP_EXPORT zsock_t *
wap_client_msgpipe (wap_client_t *self); wap_client_msgpipe (wap_client_t *self);
// Return true if client is currently connected, else false. Note that the
// client will automatically re-connect if the server dies and restarts after
// a successful first connection.
WAP_EXPORT bool
wap_client_connected (wap_client_t *self);
// Connect to server endpoint, with specified timeout in msecs (zero means wait // Connect to server endpoint, with specified timeout in msecs (zero means wait
// forever). Constructor succeeds if connection is successful. The caller may // forever). Constructor succeeds if connection is successful. The caller may
// specify its address. // specify its address.
@ -69,7 +75,7 @@ WAP_EXPORT int
// Send a raw transaction to the daemon. // Send a raw transaction to the daemon.
// Returns >= 0 if successful, -1 if interrupted. // Returns >= 0 if successful, -1 if interrupted.
WAP_EXPORT int WAP_EXPORT int
wap_client_put (wap_client_t *self, zchunk_t **tx_data_p); wap_client_put (wap_client_t *self, zchunk_t **tx_as_hex_p);
// Request a set of blocks from the server. // Request a set of blocks from the server.
// Returns >= 0 if successful, -1 if interrupted. // Returns >= 0 if successful, -1 if interrupted.
@ -86,6 +92,11 @@ WAP_EXPORT int
WAP_EXPORT int WAP_EXPORT int
wap_client_output_indexes (wap_client_t *self, const char *tx_id); wap_client_output_indexes (wap_client_t *self, const char *tx_id);
// Ask for tx output indexes.
// Returns >= 0 if successful, -1 if interrupted.
WAP_EXPORT int
wap_client_random_outs (wap_client_t *self, uint64_t outs_count, zframe_t **amounts_p);
// Send start command to server. // Send start command to server.
// Returns >= 0 if successful, -1 if interrupted. // Returns >= 0 if successful, -1 if interrupted.
WAP_EXPORT int WAP_EXPORT int
@ -124,6 +135,10 @@ WAP_EXPORT zchunk_t *
WAP_EXPORT zframe_t * WAP_EXPORT zframe_t *
wap_client_o_indexes (wap_client_t *self); wap_client_o_indexes (wap_client_t *self);
// Return last received random_outputs
WAP_EXPORT zframe_t *
wap_client_random_outputs (wap_client_t *self);
// Self test of this class // Self test of this class
WAP_EXPORT void WAP_EXPORT void
wap_client_test (bool verbose); wap_client_test (bool verbose);

404
src/ipc/include/wap_client_engine.inc
View File

@ -31,10 +31,11 @@ typedef enum {
expect_start_ok_state = 8, expect_start_ok_state = 8,
expect_stop_ok_state = 9, expect_stop_ok_state = 9,
expect_output_indexes_ok_state = 10, expect_output_indexes_ok_state = 10,
expect_close_ok_state = 11, expect_random_outs_ok_state = 11,
defaults_state = 12, expect_close_ok_state = 12,
have_error_state = 13, defaults_state = 13,
reexpect_open_ok_state = 14 have_error_state = 14,
reexpect_open_ok_state = 15
} state_t; } state_t;
typedef enum { typedef enum {
@ -50,19 +51,22 @@ typedef enum {
start_event = 9, start_event = 9,
stop_event = 10, stop_event = 10,
output_indexes_event = 11, output_indexes_event = 11,
destructor_event = 12, random_outs_event = 12,
blocks_ok_event = 13, destructor_event = 13,
get_ok_event = 14, blocks_ok_event = 14,
put_ok_event = 15, get_ok_event = 15,
save_ok_event = 16, put_ok_event = 16,
start_ok_event = 17, save_ok_event = 17,
stop_ok_event = 18, start_ok_event = 18,
output_indexes_ok_event = 19, stop_ok_event = 19,
close_ok_event = 20, output_indexes_ok_event = 20,
ping_ok_event = 21, random_outs_ok_event = 21,
error_event = 22, close_ok_event = 22,
command_invalid_event = 23, ping_ok_event = 23,
other_event = 24 error_event = 24,
exception_event = 25,
command_invalid_event = 26,
other_event = 27
} event_t; } event_t;
// Names for state machine logging and error reporting // Names for state machine logging and error reporting
@ -79,6 +83,7 @@ s_state_name [] = {
"expect start ok", "expect start ok",
"expect stop ok", "expect stop ok",
"expect output indexes ok", "expect output indexes ok",
"expect random outs ok",
"expect close ok", "expect close ok",
"defaults", "defaults",
"have error", "have error",
@ -99,6 +104,7 @@ s_event_name [] = {
"START", "START",
"STOP", "STOP",
"OUTPUT_INDEXES", "OUTPUT_INDEXES",
"RANDOM_OUTS",
"destructor", "destructor",
"BLOCKS_OK", "BLOCKS_OK",
"GET_OK", "GET_OK",
@ -107,9 +113,11 @@ s_event_name [] = {
"START_OK", "START_OK",
"STOP_OK", "STOP_OK",
"OUTPUT_INDEXES_OK", "OUTPUT_INDEXES_OK",
"RANDOM_OUTS_OK",
"CLOSE_OK", "CLOSE_OK",
"PING_OK", "PING_OK",
"ERROR", "ERROR",
"exception",
"command_invalid", "command_invalid",
"other" "other"
}; };
@ -127,8 +135,10 @@ struct _client_args_t {
char *identity; char *identity;
zlist_t *block_ids; zlist_t *block_ids;
uint64_t start_height; uint64_t start_height;
zchunk_t *tx_data; zchunk_t *tx_as_hex;
char *tx_id; char *tx_id;
uint64_t outs_count;
zframe_t *amounts;
char *address; char *address;
uint64_t thread_count; uint64_t thread_count;
}; };
@ -141,6 +151,7 @@ typedef struct {
zloop_t *loop; // Listen to pipe and dealer zloop_t *loop; // Listen to pipe and dealer
wap_proto_t *message; // Message received or sent wap_proto_t *message; // Message received or sent
client_args_t args; // Method arguments structure client_args_t args; // Method arguments structure
bool connected; // True if client is connected
bool terminated; // True if client is shutdown bool terminated; // True if client is shutdown
bool fsm_stopped; // "terminate" action called bool fsm_stopped; // "terminate" action called
size_t timeout; // inactivity timeout, msecs size_t timeout; // inactivity timeout, msecs
@ -178,7 +189,7 @@ static void
static void static void
signal_success (client_t *self); signal_success (client_t *self);
static void static void
use_heartbeat_timer (client_t *self); client_is_connected (client_t *self);
static void static void
signal_server_not_present (client_t *self); signal_server_not_present (client_t *self);
static void static void
@ -193,6 +204,10 @@ static void
prepare_start_command (client_t *self); prepare_start_command (client_t *self);
static void static void
prepare_get_output_indexes_command (client_t *self); prepare_get_output_indexes_command (client_t *self);
static void
prepare_get_random_outs_command (client_t *self);
static void
check_if_connection_is_dead (client_t *self);
static void static void
signal_have_blocks_ok (client_t *self); signal_have_blocks_ok (client_t *self);
static void static void
@ -207,6 +222,8 @@ static void
signal_have_stop_ok (client_t *self); signal_have_stop_ok (client_t *self);
static void static void
signal_have_output_indexes_ok (client_t *self); signal_have_output_indexes_ok (client_t *self);
static void
signal_have_random_outs_ok (client_t *self);
static void static void
signal_failure (client_t *self); signal_failure (client_t *self);
static void static void
@ -263,8 +280,9 @@ s_client_destroy (s_client_t **self_p)
zstr_free (&self->args.endpoint); zstr_free (&self->args.endpoint);
zstr_free (&self->args.identity); zstr_free (&self->args.identity);
zlist_destroy (&self->args.block_ids); zlist_destroy (&self->args.block_ids);
zchunk_destroy (&self->args.tx_data); zchunk_destroy (&self->args.tx_as_hex);
zstr_free (&self->args.tx_id); zstr_free (&self->args.tx_id);
zframe_destroy (&self->args.amounts);
zstr_free (&self->args.address); zstr_free (&self->args.address);
client_terminate (&self->client); client_terminate (&self->client);
wap_proto_destroy (&self->message); wap_proto_destroy (&self->message);
@ -322,9 +340,11 @@ engine_set_wakeup_event (client_t *client, size_t delay, event_t event)
} }
} }
// Set timeout for next protocol read. By default, will wait forever // Set heartbeat timeout. By default, the timeout is zero, meaning
// or until the process is interrupted. The timeout is in milliseconds. // infinite. Setting a non-zero timeout causes the state machine to
// The state machine must handle the "expired" event. // receive an "expired" event if is no incoming traffic for that many
// milliseconds. This cycles over and over until/unless the code sets
// a zero timeout. The state machine must handle the "expired" event.
static void static void
engine_set_timeout (client_t *client, size_t timeout) engine_set_timeout (client_t *client, size_t timeout)
@ -332,6 +352,10 @@ engine_set_timeout (client_t *client, size_t timeout)
if (client) { if (client) {
s_client_t *self = (s_client_t *) client; s_client_t *self = (s_client_t *) client;
self->timeout = timeout; self->timeout = timeout;
if (self->expiry_timer) {
zloop_timer_end (self->loop, self->expiry_timer);
self->expiry_timer = 0;
}
if (self->timeout) if (self->timeout)
self->expiry_timer = zloop_timer ( self->expiry_timer = zloop_timer (
self->loop, self->timeout, 1, s_client_handle_timeout, self); self->loop, self->timeout, 1, s_client_handle_timeout, self);
@ -356,6 +380,18 @@ engine_handle_socket (client_t *client, zsock_t *sock, zloop_reader_fn handler)
} }
} }
// Set connected to true/false. The client must call this if it wants
// to provide the API with the connected status.
static void
engine_set_connected (client_t *client, bool connected)
{
if (client) {
s_client_t *self = (s_client_t *) client;
self->connected = connected;
}
}
// Pedantic compilers don't like unused functions, so we call the whole // Pedantic compilers don't like unused functions, so we call the whole
// API, passing null references. It's nasty and horrid and sufficient. // API, passing null references. It's nasty and horrid and sufficient.
@ -400,6 +436,12 @@ s_protocol_event (s_client_t *self, wap_proto_t *message)
case WAP_PROTO_OUTPUT_INDEXES_OK: case WAP_PROTO_OUTPUT_INDEXES_OK:
return output_indexes_ok_event; return output_indexes_ok_event;
break; break;
case WAP_PROTO_RANDOM_OUTS:
return random_outs_event;
break;
case WAP_PROTO_RANDOM_OUTS_OK:
return random_outs_ok_event;
break;
case WAP_PROTO_GET: case WAP_PROTO_GET:
return get_event; return get_event;
break; break;
@ -526,10 +568,10 @@ s_client_execute (s_client_t *self, event_t event)
signal_success (&self->client); signal_success (&self->client);
} }
if (!self->exception) { if (!self->exception) {
// use heartbeat timer // client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ use heartbeat timer"); zsys_debug ("wap_client: $ client is connected");
use_heartbeat_timer (&self->client); client_is_connected (&self->client);
} }
if (!self->exception) if (!self->exception)
self->state = connected_state; self->state = connected_state;
@ -551,9 +593,12 @@ s_client_execute (s_client_t *self, event_t event)
} }
else else
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -566,6 +611,12 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -695,6 +746,24 @@ s_client_execute (s_client_t *self, event_t event)
self->state = expect_output_indexes_ok_state; self->state = expect_output_indexes_ok_state;
} }
else else
if (self->event == random_outs_event) {
if (!self->exception) {
// prepare get random outs command
if (wap_client_verbose)
zsys_debug ("wap_client: $ prepare get random outs command");
prepare_get_random_outs_command (&self->client);
}
if (!self->exception) {
// send RANDOM_OUTS
if (wap_client_verbose)
zsys_debug ("wap_client: $ send RANDOM_OUTS");
wap_proto_set_id (self->message, WAP_PROTO_RANDOM_OUTS);
wap_proto_send (self->message, self->dealer);
}
if (!self->exception)
self->state = expect_random_outs_ok_state;
}
else
if (self->event == destructor_event) { if (self->event == destructor_event) {
if (!self->exception) { if (!self->exception) {
// send CLOSE // send CLOSE
@ -708,6 +777,12 @@ s_client_execute (s_client_t *self, event_t event)
} }
else else
if (self->event == expired_event) { if (self->event == expired_event) {
if (!self->exception) {
// check if connection is dead
if (wap_client_verbose)
zsys_debug ("wap_client: $ check if connection is dead");
check_if_connection_is_dead (&self->client);
}
if (!self->exception) { if (!self->exception) {
// send PING // send PING
if (wap_client_verbose) if (wap_client_verbose)
@ -718,9 +793,12 @@ s_client_execute (s_client_t *self, event_t event)
} }
else else
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -733,6 +811,12 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -754,9 +838,12 @@ s_client_execute (s_client_t *self, event_t event)
} }
else else
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -769,6 +856,12 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -790,9 +883,12 @@ s_client_execute (s_client_t *self, event_t event)
} }
else else
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -805,6 +901,12 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -826,9 +928,12 @@ s_client_execute (s_client_t *self, event_t event)
} }
else else
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -841,6 +946,12 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -862,9 +973,12 @@ s_client_execute (s_client_t *self, event_t event)
} }
else else
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -877,6 +991,12 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -898,9 +1018,12 @@ s_client_execute (s_client_t *self, event_t event)
} }
else else
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -913,6 +1036,12 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -934,9 +1063,12 @@ s_client_execute (s_client_t *self, event_t event)
} }
else else
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -949,6 +1081,12 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -970,9 +1108,12 @@ s_client_execute (s_client_t *self, event_t event)
} }
else else
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -985,6 +1126,57 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else {
// Handle unexpected protocol events
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ *");
}
break;
case expect_random_outs_ok_state:
if (self->event == random_outs_ok_event) {
if (!self->exception) {
// signal have random outs ok
if (wap_client_verbose)
zsys_debug ("wap_client: $ signal have random outs ok");
signal_have_random_outs_ok (&self->client);
}
if (!self->exception)
self->state = connected_state;
}
else
if (self->event == ping_ok_event) {
if (!self->exception) {
// client is connected
if (wap_client_verbose)
zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
}
else
if (self->event == error_event) {
if (!self->exception) {
// check status code
if (wap_client_verbose)
zsys_debug ("wap_client: $ check status code");
check_status_code (&self->client);
}
if (!self->exception)
self->state = have_error_state;
}
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -1025,9 +1217,12 @@ s_client_execute (s_client_t *self, event_t event)
} }
else else
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -1040,6 +1235,12 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -1050,9 +1251,12 @@ s_client_execute (s_client_t *self, event_t event)
case defaults_state: case defaults_state:
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -1065,6 +1269,12 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -1117,19 +1327,22 @@ s_client_execute (s_client_t *self, event_t event)
case reexpect_open_ok_state: case reexpect_open_ok_state:
if (self->event == open_ok_event) { if (self->event == open_ok_event) {
if (!self->exception) { if (!self->exception) {
// use heartbeat timer // client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ use heartbeat timer"); zsys_debug ("wap_client: $ client is connected");
use_heartbeat_timer (&self->client); client_is_connected (&self->client);
} }
if (!self->exception) if (!self->exception)
self->state = connected_state; self->state = connected_state;
} }
else else
if (self->event == ping_ok_event) { if (self->event == ping_ok_event) {
// No action - just logging if (!self->exception) {
// client is connected
if (wap_client_verbose) if (wap_client_verbose)
zsys_debug ("wap_client: $ ping_ok"); zsys_debug ("wap_client: $ client is connected");
client_is_connected (&self->client);
}
} }
else else
if (self->event == error_event) { if (self->event == error_event) {
@ -1142,6 +1355,12 @@ s_client_execute (s_client_t *self, event_t event)
if (!self->exception) if (!self->exception)
self->state = have_error_state; self->state = have_error_state;
} }
else
if (self->event == exception_event) {
// No action - just logging
if (wap_client_verbose)
zsys_debug ("wap_client: $ exception");
}
else { else {
// Handle unexpected protocol events // Handle unexpected protocol events
// No action - just logging // No action - just logging
@ -1169,7 +1388,13 @@ s_client_handle_timeout (zloop_t *loop, int timer_id, void *argument)
{ {
s_client_t *self = (s_client_t *) argument; s_client_t *self = (s_client_t *) argument;
s_client_execute (self, expired_event); s_client_execute (self, expired_event);
return self->terminated? -1: 0; if (self->terminated)
return -1;
if (self->timeout > 0)
self->expiry_timer = zloop_timer (
loop, self->timeout, 1, s_client_handle_timeout, self);
return 0;
} }
// zloop callback when client wakeup timer expires // zloop callback when client wakeup timer expires
@ -1198,6 +1423,9 @@ s_client_handle_cmdpipe (zloop_t *loop, zsock_t *reader, void *argument)
if (streq (method, "$TERM")) if (streq (method, "$TERM"))
self->terminated = true; // Shutdown the engine self->terminated = true; // Shutdown the engine
else else
if (streq (method, "$CONNECTED"))
zsock_send (self->cmdpipe, "i", self->connected);
else
if (streq (method, "CONNECT")) { if (streq (method, "CONNECT")) {
zstr_free (&self->args.endpoint); zstr_free (&self->args.endpoint);
zstr_free (&self->args.identity); zstr_free (&self->args.identity);
@ -1216,8 +1444,8 @@ s_client_handle_cmdpipe (zloop_t *loop, zsock_t *reader, void *argument)
} }
else else
if (streq (method, "PUT")) { if (streq (method, "PUT")) {
zchunk_destroy (&self->args.tx_data); zchunk_destroy (&self->args.tx_as_hex);
zsock_recv (self->cmdpipe, "p", &self->args.tx_data); zsock_recv (self->cmdpipe, "p", &self->args.tx_as_hex);
s_client_execute (self, put_event); s_client_execute (self, put_event);
} }
else else
@ -1237,6 +1465,12 @@ s_client_handle_cmdpipe (zloop_t *loop, zsock_t *reader, void *argument)
s_client_execute (self, output_indexes_event); s_client_execute (self, output_indexes_event);
} }
else else
if (streq (method, "RANDOM OUTS")) {
zframe_destroy (&self->args.amounts);
zsock_recv (self->cmdpipe, "8p", &self->args.outs_count, &self->args.amounts);
s_client_execute (self, random_outs_event);
}
else
if (streq (method, "START")) { if (streq (method, "START")) {
zstr_free (&self->args.address); zstr_free (&self->args.address);
zsock_recv (self->cmdpipe, "s8", &self->args.address, &self->args.thread_count); zsock_recv (self->cmdpipe, "s8", &self->args.address, &self->args.thread_count);
@ -1357,6 +1591,7 @@ wap_client (zsock_t *cmdpipe, void *msgpipe)
struct _wap_client_t { struct _wap_client_t {
zactor_t *actor; // Client actor zactor_t *actor; // Client actor
zsock_t *msgpipe; // Pipe for async message flow zsock_t *msgpipe; // Pipe for async message flow
bool connected; // Client currently connected or not
int status; // Returned by actor reply int status; // Returned by actor reply
char *reason; // Returned by actor reply char *reason; // Returned by actor reply
uint64_t start_height; // Returned by actor reply uint64_t start_height; // Returned by actor reply
@ -1364,6 +1599,7 @@ struct _wap_client_t {
zmsg_t *block_data; // Returned by actor reply zmsg_t *block_data; // Returned by actor reply
zchunk_t *tx_data; // Returned by actor reply zchunk_t *tx_data; // Returned by actor reply
zframe_t *o_indexes; // Returned by actor reply zframe_t *o_indexes; // Returned by actor reply
zframe_t *random_outputs; // Returned by actor reply
}; };
@ -1415,6 +1651,7 @@ wap_client_destroy (wap_client_t **self_p)
zmsg_destroy (&self->block_data); zmsg_destroy (&self->block_data);
zchunk_destroy (&self->tx_data); zchunk_destroy (&self->tx_data);
zframe_destroy (&self->o_indexes); zframe_destroy (&self->o_indexes);
zframe_destroy (&self->random_outputs);
free (self); free (self);
*self_p = NULL; *self_p = NULL;
} }
@ -1448,6 +1685,22 @@ wap_client_msgpipe (wap_client_t *self)
} }
// ---------------------------------------------------------------------------
// Return true if client is currently connected, else false. Note that the
// client will automatically re-connect if the server dies and restarts after
// a successful first connection.
bool
wap_client_connected (wap_client_t *self)
{
assert (self);
bool connected;
zsock_send (self->actor, "s", "$CONNECTED");
zsock_recv (self->actor, "i", &connected);
return connected;
}
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Get valid reply from actor; discard replies that does not match. Current // Get valid reply from actor; discard replies that does not match. Current
// implementation filters on first frame of message. Blocks until a valid // implementation filters on first frame of message. Blocks until a valid
@ -1500,6 +1753,11 @@ s_accept_reply (wap_client_t *self, ...)
zsock_recv (self->actor, "8p", &self->status, &self->o_indexes); zsock_recv (self->actor, "8p", &self->status, &self->o_indexes);
} }
else else
if (streq (reply, "RANDOM OUTS OK")) {
zframe_destroy (&self->random_outputs);
zsock_recv (self->actor, "8p", &self->status, &self->random_outputs);
}
else
if (streq (reply, "START OK")) { if (streq (reply, "START OK")) {
zsock_recv (self->actor, "8", &self->status); zsock_recv (self->actor, "8", &self->status);
} }
@ -1580,12 +1838,12 @@ wap_client_blocks (wap_client_t *self, zlist_t **block_ids_p, uint64_t start_hei
// Returns >= 0 if successful, -1 if interrupted. // Returns >= 0 if successful, -1 if interrupted.
int int
wap_client_put (wap_client_t *self, zchunk_t **tx_data_p) wap_client_put (wap_client_t *self, zchunk_t **tx_as_hex_p)
{ {
assert (self); assert (self);
zsock_send (self->actor, "sp", "PUT", *tx_data_p); zsock_send (self->actor, "sp", "PUT", *tx_as_hex_p);
*tx_data_p = NULL; // Take ownership of tx_data *tx_as_hex_p = NULL; // Take ownership of tx_as_hex
if (s_accept_reply (self, "PUT OK", "FAILURE", NULL)) if (s_accept_reply (self, "PUT OK", "FAILURE", NULL))
return -1; // Interrupted or timed-out return -1; // Interrupted or timed-out
return self->status; return self->status;
@ -1640,6 +1898,23 @@ wap_client_output_indexes (wap_client_t *self, const char *tx_id)
} }
// ---------------------------------------------------------------------------
// Ask for tx output indexes.
// Returns >= 0 if successful, -1 if interrupted.
int
wap_client_random_outs (wap_client_t *self, uint64_t outs_count, zframe_t **amounts_p)
{
assert (self);
zsock_send (self->actor, "s8p", "RANDOM OUTS", outs_count, *amounts_p);
*amounts_p = NULL; // Take ownership of amounts
if (s_accept_reply (self, "RANDOM OUTS OK", "FAILURE", NULL))
return -1; // Interrupted or timed-out
return self->status;
}
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Send start command to server. // Send start command to server.
// Returns >= 0 if successful, -1 if interrupted. // Returns >= 0 if successful, -1 if interrupted.
@ -1747,3 +2022,14 @@ wap_client_o_indexes (wap_client_t *self)
assert (self); assert (self);
return self->o_indexes; return self->o_indexes;
} }
// ---------------------------------------------------------------------------
// Return last received random_outputs
zframe_t *
wap_client_random_outputs (wap_client_t *self)
{
assert (self);
return self->random_outputs;
}

86
src/ipc/include/wap_proto.h
View File

@ -43,7 +43,7 @@ BLOCKS-OK, or ERROR if the request is invalid.
PUT - Wallet sends a raw transaction to the daemon. Daemon replies with PUT - Wallet sends a raw transaction to the daemon. Daemon replies with
PUT-OK, or ERROR. PUT-OK, or ERROR.
tx_data chunk Transaction data tx_as_hex chunk Transaction as hex
PUT_OK - Daemon confirms that it accepted the raw transaction. PUT_OK - Daemon confirms that it accepted the raw transaction.
status number 8 Transaction ID status number 8 Transaction ID
@ -55,6 +55,14 @@ PUT-OK, or ERROR.
status number 8 Status status number 8 Status
o_indexes frame Output Indexes o_indexes frame Output Indexes
RANDOM_OUTS - Get random outputs for amounts.
outs_count number 8 Outs count
amounts frame Amounts
RANDOM_OUTS_OK - Daemon returns random outputs for amounts.
status number 8 Status
random_outputs frame Outputs
GET - Wallet requests transaction data from the daemon. Daemon replies GET - Wallet requests transaction data from the daemon. Daemon replies
with GET-OK, or ERROR. with GET-OK, or ERROR.
tx_id string Transaction ID tx_id string Transaction ID
@ -113,19 +121,21 @@ Daemon will reply with CLOSE-OK or ERROR.
#define WAP_PROTO_PUT_OK 6 #define WAP_PROTO_PUT_OK 6
#define WAP_PROTO_OUTPUT_INDEXES 7 #define WAP_PROTO_OUTPUT_INDEXES 7
#define WAP_PROTO_OUTPUT_INDEXES_OK 8 #define WAP_PROTO_OUTPUT_INDEXES_OK 8
#define WAP_PROTO_GET 9 #define WAP_PROTO_RANDOM_OUTS 9
#define WAP_PROTO_GET_OK 10 #define WAP_PROTO_RANDOM_OUTS_OK 10
#define WAP_PROTO_SAVE 11 #define WAP_PROTO_GET 11
#define WAP_PROTO_SAVE_OK 12 #define WAP_PROTO_GET_OK 12
#define WAP_PROTO_START 13 #define WAP_PROTO_SAVE 13
#define WAP_PROTO_START_OK 14 #define WAP_PROTO_SAVE_OK 14
#define WAP_PROTO_STOP 15 #define WAP_PROTO_START 15
#define WAP_PROTO_STOP_OK 16 #define WAP_PROTO_START_OK 16
#define WAP_PROTO_CLOSE 17 #define WAP_PROTO_STOP 17
#define WAP_PROTO_CLOSE_OK 18 #define WAP_PROTO_STOP_OK 18
#define WAP_PROTO_PING 19 #define WAP_PROTO_CLOSE 19
#define WAP_PROTO_PING_OK 20 #define WAP_PROTO_CLOSE_OK 20
#define WAP_PROTO_ERROR 21 #define WAP_PROTO_PING 21
#define WAP_PROTO_PING_OK 22
#define WAP_PROTO_ERROR 23
#include <czmq.h> #include <czmq.h>
@ -219,15 +229,15 @@ zmsg_t *
void void
wap_proto_set_block_data (wap_proto_t *self, zmsg_t **msg_p); wap_proto_set_block_data (wap_proto_t *self, zmsg_t **msg_p);
// Get a copy of the tx_data field // Get a copy of the tx_as_hex field
zchunk_t * zchunk_t *
wap_proto_tx_data (wap_proto_t *self); wap_proto_tx_as_hex (wap_proto_t *self);
// Get the tx_data field and transfer ownership to caller // Get the tx_as_hex field and transfer ownership to caller
zchunk_t * zchunk_t *
wap_proto_get_tx_data (wap_proto_t *self); wap_proto_get_tx_as_hex (wap_proto_t *self);
// Set the tx_data field, transferring ownership from caller // Set the tx_as_hex field, transferring ownership from caller
void void
wap_proto_set_tx_data (wap_proto_t *self, zchunk_t **chunk_p); wap_proto_set_tx_as_hex (wap_proto_t *self, zchunk_t **chunk_p);
// Get/set the tx_id field // Get/set the tx_id field
const char * const char *
@ -245,6 +255,42 @@ zframe_t *
void void
wap_proto_set_o_indexes (wap_proto_t *self, zframe_t **frame_p); wap_proto_set_o_indexes (wap_proto_t *self, zframe_t **frame_p);
// Get/set the outs_count field
uint64_t
wap_proto_outs_count (wap_proto_t *self);
void
wap_proto_set_outs_count (wap_proto_t *self, uint64_t outs_count);
// Get a copy of the amounts field
zframe_t *
wap_proto_amounts (wap_proto_t *self);
// Get the amounts field and transfer ownership to caller
zframe_t *
wap_proto_get_amounts (wap_proto_t *self);
// Set the amounts field, transferring ownership from caller
void
wap_proto_set_amounts (wap_proto_t *self, zframe_t **frame_p);
// Get a copy of the random_outputs field
zframe_t *
wap_proto_random_outputs (wap_proto_t *self);
// Get the random_outputs field and transfer ownership to caller
zframe_t *
wap_proto_get_random_outputs (wap_proto_t *self);
// Set the random_outputs field, transferring ownership from caller
void
wap_proto_set_random_outputs (wap_proto_t *self, zframe_t **frame_p);
// Get a copy of the tx_data field
zchunk_t *
wap_proto_tx_data (wap_proto_t *self);
// Get the tx_data field and transfer ownership to caller
zchunk_t *
wap_proto_get_tx_data (wap_proto_t *self);
// Set the tx_data field, transferring ownership from caller
void
wap_proto_set_tx_data (wap_proto_t *self, zchunk_t **chunk_p);
// Get/set the address field // Get/set the address field
const char * const char *
wap_proto_address (wap_proto_t *self); wap_proto_address (wap_proto_t *self);

37
src/ipc/include/wap_server_engine.inc
View File

@ -38,11 +38,12 @@ typedef enum {
start_event = 7, start_event = 7,
stop_event = 8, stop_event = 8,
output_indexes_event = 9, output_indexes_event = 9,
close_event = 10, random_outs_event = 10,
ping_event = 11, close_event = 11,
expired_event = 12, ping_event = 12,
exception_event = 13, expired_event = 13,
settled_event = 14 exception_event = 14,
settled_event = 15
} event_t; } event_t;
// Names for state machine logging and error reporting // Names for state machine logging and error reporting
@ -67,6 +68,7 @@ s_event_name [] = {
"START", "START",
"STOP", "STOP",
"OUTPUT_INDEXES", "OUTPUT_INDEXES",
"RANDOM_OUTS",
"CLOSE", "CLOSE",
"PING", "PING",
"expired", "expired",
@ -148,6 +150,8 @@ static void
stop_mining_process (client_t *self); stop_mining_process (client_t *self);
static void static void
output_indexes (client_t *self); output_indexes (client_t *self);
static void
random_outs (client_t *self);
static void static void
deregister_wallet (client_t *self); deregister_wallet (client_t *self);
static void static void
@ -344,6 +348,9 @@ s_protocol_event (wap_proto_t *message)
case WAP_PROTO_OUTPUT_INDEXES: case WAP_PROTO_OUTPUT_INDEXES:
return output_indexes_event; return output_indexes_event;
break; break;
case WAP_PROTO_RANDOM_OUTS:
return random_outs_event;
break;
case WAP_PROTO_GET: case WAP_PROTO_GET:
return get_event; return get_event;
break; break;
@ -691,6 +698,24 @@ s_client_execute (s_client_t *self, event_t event)
} }
} }
else else
if (self->event == random_outs_event) {
if (!self->exception) {
// random outs
if (self->server->verbose)
zsys_debug ("%s: $ random outs", self->log_prefix);
random_outs (&self->client);
}
if (!self->exception) {
// send RANDOM_OUTS_OK
if (self->server->verbose)
zsys_debug ("%s: $ send RANDOM_OUTS_OK",
self->log_prefix);
wap_proto_set_id (self->server->message, WAP_PROTO_RANDOM_OUTS_OK);
wap_proto_set_routing_id (self->server->message, self->routing_id);
wap_proto_send (self->server->message, self->server->router);
}
}
else
if (self->event == close_event) { if (self->event == close_event) {
if (!self->exception) { if (!self->exception) {
// send CLOSE_OK // send CLOSE_OK
@ -1092,6 +1117,7 @@ s_server_config_service (s_server_t *self)
if (zsock_bind (self->router, "%s", endpoint) == -1) if (zsock_bind (self->router, "%s", endpoint) == -1)
zsys_warning ("could not bind to %s (%s)", endpoint, zmq_strerror (zmq_errno ())); zsys_warning ("could not bind to %s (%s)", endpoint, zmq_strerror (zmq_errno ()));
} }
#if (ZMQ_VERSION_MAJOR >= 4)
else else
if (streq (zconfig_name (section), "security")) { if (streq (zconfig_name (section), "security")) {
char *mechanism = zconfig_resolve (section, "mechanism", "null"); char *mechanism = zconfig_resolve (section, "mechanism", "null");
@ -1109,6 +1135,7 @@ s_server_config_service (s_server_t *self)
else else
zsys_warning ("mechanism=%s is not supported", mechanism); zsys_warning ("mechanism=%s is not supported", mechanism);
} }
#endif
section = zconfig_next (section); section = zconfig_next (section);
} }
s_server_config_global (self); s_server_config_global (self);

53
src/ipc/wap_client/wap_client.c
View File

@ -37,6 +37,7 @@ typedef struct {
// Own properties // Own properties
int heartbeat_timer; // Timeout for heartbeats to server int heartbeat_timer; // Timeout for heartbeats to server
int retries; // How many heartbeats we've tried
} client_t; } client_t;
// Include the generated client engine // Include the generated client engine
@ -97,6 +98,32 @@ use_connect_timeout (client_t *self)
engine_set_timeout (self, self->args->timeout); engine_set_timeout (self, self->args->timeout);
} }
// ---------------------------------------------------------------------------
// client_is_connected
//
static void
client_is_connected (client_t *self)
{
self->retries = 0;
engine_set_connected (self, true);
engine_set_timeout (self, self->heartbeat_timer);
}
// ---------------------------------------------------------------------------
// check_if_connection_is_dead
//
static void
check_if_connection_is_dead (client_t *self)
{
// We send at most 3 heartbeats before expiring the server
if (++self->retries >= 3) {
engine_set_timeout (self, 0);
engine_set_connected (self, false);
engine_set_exception (self, exception_event);
}
}
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// use_heartbeat_timer // use_heartbeat_timer
@ -163,7 +190,7 @@ prepare_start_command (client_t *self)
static void static void
prepare_put_command (client_t *self) prepare_put_command (client_t *self)
{ {
wap_proto_set_tx_data (self->message, &self->args->tx_data); wap_proto_set_tx_as_hex (self->message, &self->args->tx_as_hex);
} }
@ -174,7 +201,7 @@ prepare_put_command (client_t *self)
static void static void
signal_have_put_ok (client_t *self) signal_have_put_ok (client_t *self)
{ {
zsock_send (self->cmdpipe, "s8s", "PUT OK", 0, zsock_send (self->cmdpipe, "s8s", "PUT OK", wap_proto_status(self->message),
wap_proto_tx_id (self->message)); wap_proto_tx_id (self->message));
} }
@ -325,3 +352,25 @@ signal_server_not_present (client_t *self)
zsock_send (self->cmdpipe, "sis", "FAILURE", -1, "Server is not reachable"); zsock_send (self->cmdpipe, "sis", "FAILURE", -1, "Server is not reachable");
} }
// ---------------------------------------------------------------------------
// prepare_get_random_outs_command
//
static void
prepare_get_random_outs_command (client_t *self)
{
wap_proto_set_amounts (self->message, &self->args->amounts);
}
// ---------------------------------------------------------------------------
// signal_have_random_outs_ok
//
static void
signal_have_random_outs_ok (client_t *self)
{
zsock_send (self->cmdpipe, "s8p", "RANDOM OUTS OK",
wap_proto_status (self->message),
wap_proto_get_random_outputs (self->message));
}

343
src/ipc/wap_proto.c
View File

@ -34,29 +34,37 @@ struct _wap_proto_t {
int id; // wap_proto message ID int id; // wap_proto message ID
byte *needle; // Read/write pointer for serialization byte *needle; // Read/write pointer for serialization
byte *ceiling; // Valid upper limit for read pointer byte *ceiling; // Valid upper limit for read pointer
/* Wallet identity */ // Wallet identity
char identity [256]; char identity [256];
/* */ // block_ids
zlist_t *block_ids; zlist_t *block_ids;
/* */ // start_height
uint64_t start_height; uint64_t start_height;
/* */ // status
uint64_t status; uint64_t status;
/* */ // curr_height
uint64_t curr_height; uint64_t curr_height;
/* Frames of block data */ // Frames of block data
zmsg_t *block_data; zmsg_t *block_data;
/* Transaction data */ // Transaction as hex
zchunk_t *tx_data; zchunk_t *tx_as_hex;
/* Transaction ID */ // Transaction ID
char tx_id [256]; char tx_id [256];
/* Output Indexes */ // Output Indexes
zframe_t *o_indexes; zframe_t *o_indexes;
/* */ // Outs count
uint64_t outs_count;
// Amounts
zframe_t *amounts;
// Outputs
zframe_t *random_outputs;
// Transaction data
zchunk_t *tx_data;
// address
char address [256]; char address [256];
/* */ // thread_count
uint64_t thread_count; uint64_t thread_count;
/* Printable explanation */ // Printable explanation
char reason [256]; char reason [256];
}; };
@ -236,8 +244,11 @@ wap_proto_destroy (wap_proto_t **self_p)
if (self->block_ids) if (self->block_ids)
zlist_destroy (&self->block_ids); zlist_destroy (&self->block_ids);
zmsg_destroy (&self->block_data); zmsg_destroy (&self->block_data);
zchunk_destroy (&self->tx_data); zchunk_destroy (&self->tx_as_hex);
zframe_destroy (&self->o_indexes); zframe_destroy (&self->o_indexes);
zframe_destroy (&self->amounts);
zframe_destroy (&self->random_outputs);
zchunk_destroy (&self->tx_data);
// Free object itself // Free object itself
free (self); free (self);
@ -342,11 +353,11 @@ wap_proto_recv (wap_proto_t *self, zsock_t *input)
size_t chunk_size; size_t chunk_size;
GET_NUMBER4 (chunk_size); GET_NUMBER4 (chunk_size);
if (self->needle + chunk_size > (self->ceiling)) { if (self->needle + chunk_size > (self->ceiling)) {
zsys_warning ("wap_proto: tx_data is missing data"); zsys_warning ("wap_proto: tx_as_hex is missing data");
goto malformed; goto malformed;
} }
zchunk_destroy (&self->tx_data); zchunk_destroy (&self->tx_as_hex);
self->tx_data = zchunk_new (self->needle, chunk_size); self->tx_as_hex = zchunk_new (self->needle, chunk_size);
self->needle += chunk_size; self->needle += chunk_size;
} }
break; break;
@ -370,6 +381,28 @@ wap_proto_recv (wap_proto_t *self, zsock_t *input)
self->o_indexes = zframe_recv (input); self->o_indexes = zframe_recv (input);
break; break;
case WAP_PROTO_RANDOM_OUTS:
GET_NUMBER8 (self->outs_count);
// Get next frame off socket
if (!zsock_rcvmore (input)) {
zsys_warning ("wap_proto: amounts is missing");
goto malformed;
}
zframe_destroy (&self->amounts);
self->amounts = zframe_recv (input);
break;
case WAP_PROTO_RANDOM_OUTS_OK:
GET_NUMBER8 (self->status);
// Get next frame off socket
if (!zsock_rcvmore (input)) {
zsys_warning ("wap_proto: random_outputs is missing");
goto malformed;
}
zframe_destroy (&self->random_outputs);
self->random_outputs = zframe_recv (input);
break;
case WAP_PROTO_GET: case WAP_PROTO_GET:
GET_STRING (self->tx_id); GET_STRING (self->tx_id);
break; break;
@ -480,8 +513,8 @@ wap_proto_send (wap_proto_t *self, zsock_t *output)
break; break;
case WAP_PROTO_PUT: case WAP_PROTO_PUT:
frame_size += 4; // Size is 4 octets frame_size += 4; // Size is 4 octets
if (self->tx_data) if (self->tx_as_hex)
frame_size += zchunk_size (self->tx_data); frame_size += zchunk_size (self->tx_as_hex);
break; break;
case WAP_PROTO_PUT_OK: case WAP_PROTO_PUT_OK:
frame_size += 8; // status frame_size += 8; // status
@ -492,6 +525,12 @@ wap_proto_send (wap_proto_t *self, zsock_t *output)
case WAP_PROTO_OUTPUT_INDEXES_OK: case WAP_PROTO_OUTPUT_INDEXES_OK:
frame_size += 8; // status frame_size += 8; // status
break; break;
case WAP_PROTO_RANDOM_OUTS:
frame_size += 8; // outs_count
break;
case WAP_PROTO_RANDOM_OUTS_OK:
frame_size += 8; // status
break;
case WAP_PROTO_GET: case WAP_PROTO_GET:
frame_size += 1 + strlen (self->tx_id); frame_size += 1 + strlen (self->tx_id);
break; break;
@ -551,12 +590,12 @@ wap_proto_send (wap_proto_t *self, zsock_t *output)
break; break;
case WAP_PROTO_PUT: case WAP_PROTO_PUT:
if (self->tx_data) { if (self->tx_as_hex) {
PUT_NUMBER4 (zchunk_size (self->tx_data)); PUT_NUMBER4 (zchunk_size (self->tx_as_hex));
memcpy (self->needle, memcpy (self->needle,
zchunk_data (self->tx_data), zchunk_data (self->tx_as_hex),
zchunk_size (self->tx_data)); zchunk_size (self->tx_as_hex));
self->needle += zchunk_size (self->tx_data); self->needle += zchunk_size (self->tx_as_hex);
} }
else else
PUT_NUMBER4 (0); // Empty chunk PUT_NUMBER4 (0); // Empty chunk
@ -575,6 +614,16 @@ wap_proto_send (wap_proto_t *self, zsock_t *output)
nbr_frames++; nbr_frames++;
break; break;
case WAP_PROTO_RANDOM_OUTS:
PUT_NUMBER8 (self->outs_count);
nbr_frames++;
break;
case WAP_PROTO_RANDOM_OUTS_OK:
PUT_NUMBER8 (self->status);
nbr_frames++;
break;
case WAP_PROTO_GET: case WAP_PROTO_GET:
PUT_STRING (self->tx_id); PUT_STRING (self->tx_id);
break; break;
@ -617,6 +666,22 @@ wap_proto_send (wap_proto_t *self, zsock_t *output)
else else
zmq_send (zsock_resolve (output), NULL, 0, (--nbr_frames? ZMQ_SNDMORE: 0)); zmq_send (zsock_resolve (output), NULL, 0, (--nbr_frames? ZMQ_SNDMORE: 0));
} }
// Now send any frame fields, in order
if (self->id == WAP_PROTO_RANDOM_OUTS) {
// If amounts isn't set, send an empty frame
if (self->amounts)
zframe_send (&self->amounts, output, ZFRAME_REUSE + (--nbr_frames? ZFRAME_MORE: 0));
else
zmq_send (zsock_resolve (output), NULL, 0, (--nbr_frames? ZMQ_SNDMORE: 0));
}
// Now send any frame fields, in order
if (self->id == WAP_PROTO_RANDOM_OUTS_OK) {
// If random_outputs isn't set, send an empty frame
if (self->random_outputs)
zframe_send (&self->random_outputs, output, ZFRAME_REUSE + (--nbr_frames? ZFRAME_MORE: 0));
else
zmq_send (zsock_resolve (output), NULL, 0, (--nbr_frames? ZMQ_SNDMORE: 0));
}
// Now send the block_data if necessary // Now send the block_data if necessary
if (send_block_data) { if (send_block_data) {
if (self->block_data) { if (self->block_data) {
@ -682,7 +747,7 @@ wap_proto_print (wap_proto_t *self)
case WAP_PROTO_PUT: case WAP_PROTO_PUT:
zsys_debug ("WAP_PROTO_PUT:"); zsys_debug ("WAP_PROTO_PUT:");
zsys_debug (" tx_data=[ ... ]"); zsys_debug (" tx_as_hex=[ ... ]");
break; break;
case WAP_PROTO_PUT_OK: case WAP_PROTO_PUT_OK:
@ -708,6 +773,26 @@ wap_proto_print (wap_proto_t *self)
zsys_debug ("(NULL)"); zsys_debug ("(NULL)");
break; break;
case WAP_PROTO_RANDOM_OUTS:
zsys_debug ("WAP_PROTO_RANDOM_OUTS:");
zsys_debug (" outs_count=%ld", (long) self->outs_count);
zsys_debug (" amounts=");
if (self->amounts)
zframe_print (self->amounts, NULL);
else
zsys_debug ("(NULL)");
break;
case WAP_PROTO_RANDOM_OUTS_OK:
zsys_debug ("WAP_PROTO_RANDOM_OUTS_OK:");
zsys_debug (" status=%ld", (long) self->status);
zsys_debug (" random_outputs=");
if (self->random_outputs)
zframe_print (self->random_outputs, NULL);
else
zsys_debug ("(NULL)");
break;
case WAP_PROTO_GET: case WAP_PROTO_GET:
zsys_debug ("WAP_PROTO_GET:"); zsys_debug ("WAP_PROTO_GET:");
if (self->tx_id) if (self->tx_id)
@ -847,6 +932,12 @@ wap_proto_command (wap_proto_t *self)
case WAP_PROTO_OUTPUT_INDEXES_OK: case WAP_PROTO_OUTPUT_INDEXES_OK:
return ("OUTPUT_INDEXES_OK"); return ("OUTPUT_INDEXES_OK");
break; break;
case WAP_PROTO_RANDOM_OUTS:
return ("RANDOM_OUTS");
break;
case WAP_PROTO_RANDOM_OUTS_OK:
return ("RANDOM_OUTS_OK");
break;
case WAP_PROTO_GET: case WAP_PROTO_GET:
return ("GET"); return ("GET");
break; break;
@ -1035,34 +1126,34 @@ wap_proto_set_block_data (wap_proto_t *self, zmsg_t **msg_p)
// -------------------------------------------------------------------------- // --------------------------------------------------------------------------
// Get the tx_data field without transferring ownership // Get the tx_as_hex field without transferring ownership
zchunk_t * zchunk_t *
wap_proto_tx_data (wap_proto_t *self) wap_proto_tx_as_hex (wap_proto_t *self)
{ {
assert (self); assert (self);
return self->tx_data; return self->tx_as_hex;
} }
// Get the tx_data field and transfer ownership to caller // Get the tx_as_hex field and transfer ownership to caller
zchunk_t * zchunk_t *
wap_proto_get_tx_data (wap_proto_t *self) wap_proto_get_tx_as_hex (wap_proto_t *self)
{ {
zchunk_t *tx_data = self->tx_data; zchunk_t *tx_as_hex = self->tx_as_hex;
self->tx_data = NULL; self->tx_as_hex = NULL;
return tx_data; return tx_as_hex;
} }
// Set the tx_data field, transferring ownership from caller // Set the tx_as_hex field, transferring ownership from caller
void void
wap_proto_set_tx_data (wap_proto_t *self, zchunk_t **chunk_p) wap_proto_set_tx_as_hex (wap_proto_t *self, zchunk_t **chunk_p)
{ {
assert (self); assert (self);
assert (chunk_p); assert (chunk_p);
zchunk_destroy (&self->tx_data); zchunk_destroy (&self->tx_as_hex);
self->tx_data = *chunk_p; self->tx_as_hex = *chunk_p;
*chunk_p = NULL; *chunk_p = NULL;
} }
@ -1122,6 +1213,123 @@ wap_proto_set_o_indexes (wap_proto_t *self, zframe_t **frame_p)
} }
// --------------------------------------------------------------------------
// Get/set the outs_count field
uint64_t
wap_proto_outs_count (wap_proto_t *self)
{
assert (self);
return self->outs_count;
}
void
wap_proto_set_outs_count (wap_proto_t *self, uint64_t outs_count)
{
assert (self);
self->outs_count = outs_count;
}
// --------------------------------------------------------------------------
// Get the amounts field without transferring ownership
zframe_t *
wap_proto_amounts (wap_proto_t *self)
{
assert (self);
return self->amounts;
}
// Get the amounts field and transfer ownership to caller
zframe_t *
wap_proto_get_amounts (wap_proto_t *self)
{
zframe_t *amounts = self->amounts;
self->amounts = NULL;
return amounts;
}
// Set the amounts field, transferring ownership from caller
void
wap_proto_set_amounts (wap_proto_t *self, zframe_t **frame_p)
{
assert (self);
assert (frame_p);
zframe_destroy (&self->amounts);
self->amounts = *frame_p;
*frame_p = NULL;
}
// --------------------------------------------------------------------------
// Get the random_outputs field without transferring ownership
zframe_t *
wap_proto_random_outputs (wap_proto_t *self)
{
assert (self);
return self->random_outputs;
}
// Get the random_outputs field and transfer ownership to caller
zframe_t *
wap_proto_get_random_outputs (wap_proto_t *self)
{
zframe_t *random_outputs = self->random_outputs;
self->random_outputs = NULL;
return random_outputs;
}
// Set the random_outputs field, transferring ownership from caller
void
wap_proto_set_random_outputs (wap_proto_t *self, zframe_t **frame_p)
{
assert (self);
assert (frame_p);
zframe_destroy (&self->random_outputs);
self->random_outputs = *frame_p;
*frame_p = NULL;
}
// --------------------------------------------------------------------------
// Get the tx_data field without transferring ownership
zchunk_t *
wap_proto_tx_data (wap_proto_t *self)
{
assert (self);
return self->tx_data;
}
// Get the tx_data field and transfer ownership to caller
zchunk_t *
wap_proto_get_tx_data (wap_proto_t *self)
{
zchunk_t *tx_data = self->tx_data;
self->tx_data = NULL;
return tx_data;
}
// Set the tx_data field, transferring ownership from caller
void
wap_proto_set_tx_data (wap_proto_t *self, zchunk_t **chunk_p)
{
assert (self);
assert (chunk_p);
zchunk_destroy (&self->tx_data);
self->tx_data = *chunk_p;
*chunk_p = NULL;
}
// -------------------------------------------------------------------------- // --------------------------------------------------------------------------
// Get/set the address field // Get/set the address field
@ -1203,13 +1411,16 @@ wap_proto_test (bool verbose)
wap_proto_destroy (&self); wap_proto_destroy (&self);
// Create pair of sockets we can send through // Create pair of sockets we can send through
zsock_t *input = zsock_new (ZMQ_ROUTER); // We must bind before connect if we wish to remain compatible with ZeroMQ < v4
assert (input);
zsock_connect (input, "inproc://selftest-wap_proto");
zsock_t *output = zsock_new (ZMQ_DEALER); zsock_t *output = zsock_new (ZMQ_DEALER);
assert (output); assert (output);
zsock_bind (output, "inproc://selftest-wap_proto"); int rc = zsock_bind (output, "inproc://selftest-wap_proto");
assert (rc == 0);
zsock_t *input = zsock_new (ZMQ_ROUTER);
assert (input);
rc = zsock_connect (input, "inproc://selftest-wap_proto");
assert (rc == 0);
// Encode/send/decode and verify each message type // Encode/send/decode and verify each message type
int instance; int instance;
@ -1255,6 +1466,7 @@ wap_proto_test (bool verbose)
assert (streq ((char *) zlist_first (block_ids), "Name: Brutus")); assert (streq ((char *) zlist_first (block_ids), "Name: Brutus"));
assert (streq ((char *) zlist_next (block_ids), "Age: 43")); assert (streq ((char *) zlist_next (block_ids), "Age: 43"));
zlist_destroy (&block_ids); zlist_destroy (&block_ids);
zlist_destroy (&blocks_block_ids);
assert (wap_proto_start_height (self) == 123); assert (wap_proto_start_height (self) == 123);
} }
wap_proto_set_id (self, WAP_PROTO_BLOCKS_OK); wap_proto_set_id (self, WAP_PROTO_BLOCKS_OK);
@ -1264,7 +1476,7 @@ wap_proto_test (bool verbose)
wap_proto_set_curr_height (self, 123); wap_proto_set_curr_height (self, 123);
zmsg_t *blocks_ok_block_data = zmsg_new (); zmsg_t *blocks_ok_block_data = zmsg_new ();
wap_proto_set_block_data (self, &blocks_ok_block_data); wap_proto_set_block_data (self, &blocks_ok_block_data);
zmsg_addstr (wap_proto_block_data (self), "Hello, World"); zmsg_addstr (wap_proto_block_data (self), "Captcha Diem");
// Send twice // Send twice
wap_proto_send (self, output); wap_proto_send (self, output);
wap_proto_send (self, output); wap_proto_send (self, output);
@ -1276,11 +1488,15 @@ wap_proto_test (bool verbose)
assert (wap_proto_start_height (self) == 123); assert (wap_proto_start_height (self) == 123);
assert (wap_proto_curr_height (self) == 123); assert (wap_proto_curr_height (self) == 123);
assert (zmsg_size (wap_proto_block_data (self)) == 1); assert (zmsg_size (wap_proto_block_data (self)) == 1);
char *content = zmsg_popstr (wap_proto_block_data (self));
assert (streq (content, "Captcha Diem"));
zstr_free (&content);
zmsg_destroy (&blocks_ok_block_data);
} }
wap_proto_set_id (self, WAP_PROTO_PUT); wap_proto_set_id (self, WAP_PROTO_PUT);
zchunk_t *put_tx_data = zchunk_new ("Captcha Diem", 12); zchunk_t *put_tx_as_hex = zchunk_new ("Captcha Diem", 12);
wap_proto_set_tx_data (self, &put_tx_data); wap_proto_set_tx_as_hex (self, &put_tx_as_hex);
// Send twice // Send twice
wap_proto_send (self, output); wap_proto_send (self, output);
wap_proto_send (self, output); wap_proto_send (self, output);
@ -1288,7 +1504,8 @@ wap_proto_test (bool verbose)
for (instance = 0; instance < 2; instance++) { for (instance = 0; instance < 2; instance++) {
wap_proto_recv (self, input); wap_proto_recv (self, input);
assert (wap_proto_routing_id (self)); assert (wap_proto_routing_id (self));
assert (memcmp (zchunk_data (wap_proto_tx_data (self)), "Captcha Diem", 12) == 0); assert (memcmp (zchunk_data (wap_proto_tx_as_hex (self)), "Captcha Diem", 12) == 0);
zchunk_destroy (&put_tx_as_hex);
} }
wap_proto_set_id (self, WAP_PROTO_PUT_OK); wap_proto_set_id (self, WAP_PROTO_PUT_OK);
@ -1328,6 +1545,39 @@ wap_proto_test (bool verbose)
assert (wap_proto_routing_id (self)); assert (wap_proto_routing_id (self));
assert (wap_proto_status (self) == 123); assert (wap_proto_status (self) == 123);
assert (zframe_streq (wap_proto_o_indexes (self), "Captcha Diem")); assert (zframe_streq (wap_proto_o_indexes (self), "Captcha Diem"));
zframe_destroy (&output_indexes_ok_o_indexes);
}
wap_proto_set_id (self, WAP_PROTO_RANDOM_OUTS);
wap_proto_set_outs_count (self, 123);
zframe_t *random_outs_amounts = zframe_new ("Captcha Diem", 12);
wap_proto_set_amounts (self, &random_outs_amounts);
// Send twice
wap_proto_send (self, output);
wap_proto_send (self, output);
for (instance = 0; instance < 2; instance++) {
wap_proto_recv (self, input);
assert (wap_proto_routing_id (self));
assert (wap_proto_outs_count (self) == 123);
assert (zframe_streq (wap_proto_amounts (self), "Captcha Diem"));
zframe_destroy (&random_outs_amounts);
}
wap_proto_set_id (self, WAP_PROTO_RANDOM_OUTS_OK);
wap_proto_set_status (self, 123);
zframe_t *random_outs_ok_random_outputs = zframe_new ("Captcha Diem", 12);
wap_proto_set_random_outputs (self, &random_outs_ok_random_outputs);
// Send twice
wap_proto_send (self, output);
wap_proto_send (self, output);
for (instance = 0; instance < 2; instance++) {
wap_proto_recv (self, input);
assert (wap_proto_routing_id (self));
assert (wap_proto_status (self) == 123);
assert (zframe_streq (wap_proto_random_outputs (self), "Captcha Diem"));
zframe_destroy (&random_outs_ok_random_outputs);
} }
wap_proto_set_id (self, WAP_PROTO_GET); wap_proto_set_id (self, WAP_PROTO_GET);
@ -1353,6 +1603,7 @@ wap_proto_test (bool verbose)
wap_proto_recv (self, input); wap_proto_recv (self, input);
assert (wap_proto_routing_id (self)); assert (wap_proto_routing_id (self));
assert (memcmp (zchunk_data (wap_proto_tx_data (self)), "Captcha Diem", 12) == 0); assert (memcmp (zchunk_data (wap_proto_tx_data (self)), "Captcha Diem", 12) == 0);
zchunk_destroy (&get_ok_tx_data);
} }
wap_proto_set_id (self, WAP_PROTO_SAVE); wap_proto_set_id (self, WAP_PROTO_SAVE);

10
src/ipc/wap_server/wap_server.c
View File

@ -261,3 +261,13 @@ signal_command_not_valid (client_t *self)
{ {
wap_proto_set_status (self->message, WAP_PROTO_COMMAND_INVALID); wap_proto_set_status (self->message, WAP_PROTO_COMMAND_INVALID);
} }
// ---------------------------------------------------------------------------
// random_outs
//
static void
random_outs (client_t *self)
{
IPC::Daemon::get_random_outs(self->message);
}

13
src/simplewallet/simplewallet.cpp
View File

@ -1065,9 +1065,10 @@ bool simple_wallet::show_blockchain_height(const std::vector<std::string>& args)
//---------------------------------------------------------------------------------------------------- //----------------------------------------------------------------------------------------------------
bool simple_wallet::transfer(const std::vector<std::string> &args_) bool simple_wallet::transfer(const std::vector<std::string> &args_)
{ {
if (!try_connect_to_daemon()) /*if (!try_connect_to_daemon())
return true; return true;*/
std::cout << "1\n";
std::vector<std::string> local_args = args_; std::vector<std::string> local_args = args_;
if(local_args.size() < 3) if(local_args.size() < 3)
{ {
@ -1075,6 +1076,7 @@ bool simple_wallet::transfer(const std::vector<std::string> &args_)
return true; return true;
} }
std::cout << "2\n";
size_t fake_outs_count; size_t fake_outs_count;
if(!epee::string_tools::get_xtype_from_string(fake_outs_count, local_args[0])) if(!epee::string_tools::get_xtype_from_string(fake_outs_count, local_args[0]))
{ {
@ -1083,6 +1085,7 @@ bool simple_wallet::transfer(const std::vector<std::string> &args_)
} }
local_args.erase(local_args.begin()); local_args.erase(local_args.begin());
std::cout << "3\n";
std::vector<uint8_t> extra; std::vector<uint8_t> extra;
if (1 == local_args.size() % 2) if (1 == local_args.size() % 2)
{ {
@ -1105,6 +1108,7 @@ bool simple_wallet::transfer(const std::vector<std::string> &args_)
} }
} }
std::cout << "4\n";
vector<cryptonote::tx_destination_entry> dsts; vector<cryptonote::tx_destination_entry> dsts;
for (size_t i = 0; i < local_args.size(); i += 2) for (size_t i = 0; i < local_args.size(); i += 2)
{ {
@ -1179,14 +1183,17 @@ bool simple_wallet::transfer(const std::vector<std::string> &args_)
dsts.push_back(de); dsts.push_back(de);
} }
std::cout << "5\n";
try try
{ {
// figure out what tx will be necessary // figure out what tx will be necessary
auto ptx_vector = m_wallet->create_transactions(dsts, fake_outs_count, 0 /* unlock_time */, 0 /* unused fee arg*/, extra); auto ptx_vector = m_wallet->create_transactions(dsts, fake_outs_count, 0 /* unlock_time */, 0 /* unused fee arg*/, extra);
std::cout << "5a\n";
// if more than one tx necessary, prompt user to confirm // if more than one tx necessary, prompt user to confirm
if (ptx_vector.size() > 1) if (ptx_vector.size() > 1)
{ {
std::cout << "5b\n";
std::string prompt_str = "Your transaction needs to be split into "; std::string prompt_str = "Your transaction needs to be split into ";
prompt_str += std::to_string(ptx_vector.size()); prompt_str += std::to_string(ptx_vector.size());
prompt_str += " transactions. This will result in a transaction fee being applied to each transaction"; prompt_str += " transactions. This will result in a transaction fee being applied to each transaction";
@ -1202,6 +1209,7 @@ bool simple_wallet::transfer(const std::vector<std::string> &args_)
} }
} }
std::cout << "6\n";
// actually commit the transactions // actually commit the transactions
while (!ptx_vector.empty()) while (!ptx_vector.empty())
{ {
@ -1219,6 +1227,7 @@ bool simple_wallet::transfer(const std::vector<std::string> &args_)
} }
catch (const tools::error::no_connection_to_daemon&) catch (const tools::error::no_connection_to_daemon&)
{ {
std::cout << "7\n";
fail_msg_writer() << "no connection to daemon. Please, make sure daemon is running."; fail_msg_writer() << "no connection to daemon. Please, make sure daemon is running.";
} }
catch (const tools::error::wallet_rpc_error& e) catch (const tools::error::wallet_rpc_error& e)

16
src/wallet/wallet2.cpp
View File

@ -1130,13 +1130,21 @@ std::string wallet2::address_from_txt_record(const std::string& s)
void wallet2::commit_tx(pending_tx& ptx) void wallet2::commit_tx(pending_tx& ptx)
{ {
using namespace cryptonote; using namespace cryptonote;
COMMAND_RPC_SEND_RAW_TX::request req; /*COMMAND_RPC_SEND_RAW_TX::request req;
req.tx_as_hex = epee::string_tools::buff_to_hex_nodelimer(tx_to_blob(ptx.tx)); req.tx_as_hex = epee::string_tools::buff_to_hex_nodelimer(tx_to_blob(ptx.tx));
COMMAND_RPC_SEND_RAW_TX::response daemon_send_resp; COMMAND_RPC_SEND_RAW_TX::response daemon_send_resp;
bool r = epee::net_utils::invoke_http_json_remote_command2(m_daemon_address + "/sendrawtransaction", req, daemon_send_resp, m_http_client, 200000); bool r = epee::net_utils::invoke_http_json_remote_command2(m_daemon_address + "/sendrawtransaction", req, daemon_send_resp, m_http_client, 200000);
THROW_WALLET_EXCEPTION_IF(!r, error::no_connection_to_daemon, "sendrawtransaction"); THROW_WALLET_EXCEPTION_IF(!r, error::no_connection_to_daemon, "sendrawtransaction");
THROW_WALLET_EXCEPTION_IF(daemon_send_resp.status == CORE_RPC_STATUS_BUSY, error::daemon_busy, "sendrawtransaction"); THROW_WALLET_EXCEPTION_IF(daemon_send_resp.status == CORE_RPC_STATUS_BUSY, error::daemon_busy, "sendrawtransaction");
THROW_WALLET_EXCEPTION_IF(daemon_send_resp.status != CORE_RPC_STATUS_OK, error::tx_rejected, ptx.tx, daemon_send_resp.status); THROW_WALLET_EXCEPTION_IF(daemon_send_resp.status != CORE_RPC_STATUS_OK, error::tx_rejected, ptx.tx, daemon_send_resp.status);*/
std::string tx_as_hex_string = epee::string_tools::buff_to_hex_nodelimer(tx_to_blob(ptx.tx));
zchunk_t *tx_as_hex = zchunk_new((void*)tx_as_hex_string.c_str(), tx_as_hex_string.length());
int rc = wap_client_put(client, &tx_as_hex);
uint64_t status = wap_client_status(client);
THROW_WALLET_EXCEPTION_IF(status == IPC::STATUS_CORE_BUSY, error::daemon_busy, "sendrawtransaction");
THROW_WALLET_EXCEPTION_IF((status == IPC::STATUS_INVALID_TX) || (status == IPC::STATUS_TX_VERIFICATION_FAILED) ||
(status == IPC::STATUS_TX_NOT_RELAYED), error::tx_rejected, ptx.tx, status);
add_unconfirmed_tx(ptx.tx, ptx.change_dts.amount); add_unconfirmed_tx(ptx.tx, ptx.change_dts.amount);
@ -1171,6 +1179,7 @@ std::vector<wallet2::pending_tx> wallet2::create_transactions(std::vector<crypto
// failsafe split attempt counter // failsafe split attempt counter
size_t attempt_count = 0; size_t attempt_count = 0;
std::cout << "a\n";
for(attempt_count = 1; ;attempt_count++) for(attempt_count = 1; ;attempt_count++)
{ {
auto split_values = split_amounts(dsts, attempt_count); auto split_values = split_amounts(dsts, attempt_count);
@ -1190,11 +1199,14 @@ std::vector<wallet2::pending_tx> wallet2::create_transactions(std::vector<crypto
cryptonote::transaction tx; cryptonote::transaction tx;
pending_tx ptx; pending_tx ptx;
std::cout << "b\n";
// loop until fee is met without increasing tx size to next KB boundary. // loop until fee is met without increasing tx size to next KB boundary.
uint64_t needed_fee = 0; uint64_t needed_fee = 0;
do do
{ {
std::cout << "c\n";
transfer(dst_vector, fake_outs_count, unlock_time, needed_fee, extra, tx, ptx); transfer(dst_vector, fake_outs_count, unlock_time, needed_fee, extra, tx, ptx);
std::cout << "d\n";
auto txBlob = t_serializable_object_to_blob(ptx.tx); auto txBlob = t_serializable_object_to_blob(ptx.tx);
uint64_t txSize = txBlob.size(); uint64_t txSize = txBlob.size();
uint64_t numKB = txSize / 1024; uint64_t numKB = txSize / 1024;

41
src/wallet/wallet2.h
View File

@ -471,23 +471,54 @@ namespace tools
COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::response daemon_resp = AUTO_VAL_INIT(daemon_resp); COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::response daemon_resp = AUTO_VAL_INIT(daemon_resp);
if(fake_outputs_count) if(fake_outputs_count)
{ {
COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::request req = AUTO_VAL_INIT(req); // COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::request req = AUTO_VAL_INIT(req);
req.outs_count = fake_outputs_count + 1;// add one to make possible (if need) to skip real output key // req.outs_count = fake_outputs_count + 1;// add one to make possible (if need) to skip real output key
uint64_t outs_count = fake_outputs_count + 1;
std::vector<uint64_t> amounts;
BOOST_FOREACH(transfer_container::iterator it, selected_transfers) BOOST_FOREACH(transfer_container::iterator it, selected_transfers)
{ {
THROW_WALLET_EXCEPTION_IF(it->m_tx.vout.size() <= it->m_internal_output_index, error::wallet_internal_error, THROW_WALLET_EXCEPTION_IF(it->m_tx.vout.size() <= it->m_internal_output_index, error::wallet_internal_error,
"m_internal_output_index = " + std::to_string(it->m_internal_output_index) + "m_internal_output_index = " + std::to_string(it->m_internal_output_index) +
" is greater or equal to outputs count = " + std::to_string(it->m_tx.vout.size())); " is greater or equal to outputs count = " + std::to_string(it->m_tx.vout.size()));
req.amounts.push_back(it->amount()); amounts.push_back(it->amount());
} }
bool r = epee::net_utils::invoke_http_bin_remote_command2(m_daemon_address + "/getrandom_outs.bin", req, daemon_resp, m_http_client, 200000); zframe_t *amounts_frame = zframe_new(&amounts[0], amounts.size() * sizeof(uint64_t));
int rc = wap_client_random_outs(client, outs_count, &amounts_frame);
/*bool r = epee::net_utils::invoke_http_bin_remote_command2(m_daemon_address + "/getrandom_outs.bin", req, daemon_resp, m_http_client, 200000);
THROW_WALLET_EXCEPTION_IF(!r, error::no_connection_to_daemon, "getrandom_outs.bin"); THROW_WALLET_EXCEPTION_IF(!r, error::no_connection_to_daemon, "getrandom_outs.bin");
THROW_WALLET_EXCEPTION_IF(daemon_resp.status == CORE_RPC_STATUS_BUSY, error::daemon_busy, "getrandom_outs.bin"); THROW_WALLET_EXCEPTION_IF(daemon_resp.status == CORE_RPC_STATUS_BUSY, error::daemon_busy, "getrandom_outs.bin");
THROW_WALLET_EXCEPTION_IF(daemon_resp.status != CORE_RPC_STATUS_OK, error::get_random_outs_error, daemon_resp.status); THROW_WALLET_EXCEPTION_IF(daemon_resp.status != CORE_RPC_STATUS_OK, error::get_random_outs_error, daemon_resp.status);
THROW_WALLET_EXCEPTION_IF(daemon_resp.outs.size() != selected_transfers.size(), error::wallet_internal_error, THROW_WALLET_EXCEPTION_IF(daemon_resp.outs.size() != selected_transfers.size(), error::wallet_internal_error,
"daemon returned wrong response for getrandom_outs.bin, wrong amounts count = " + "daemon returned wrong response for getrandom_outs.bin, wrong amounts count = " +
std::to_string(daemon_resp.outs.size()) + ", expected " + std::to_string(selected_transfers.size())); std::to_string(daemon_resp.outs.size()) + ", expected " + std::to_string(selected_transfers.size()));*/
uint64_t status = wap_client_status(client);
THROW_WALLET_EXCEPTION_IF(status == IPC::STATUS_CORE_BUSY, error::daemon_busy, "getrandomouts");
// TODO: Use a code to string mapping of errors
THROW_WALLET_EXCEPTION_IF(status == IPC::STATUS_RANDOM_OUTS_FAILED, error::get_random_outs_error, "IPC::STATUS_RANDOM_OUTS_FAILED");
THROW_WALLET_EXCEPTION_IF(status != IPC::STATUS_OK, error::get_random_outs_error, "!IPC:STATUS_OK");
// Convert ZMQ response back into RPC response object.
zframe_t *outputs_frame = wap_client_random_outputs(client);
uint64_t frame_size = zframe_size(outputs_frame);
char *frame_data = reinterpret_cast<char*>(zframe_data(outputs_frame));
std::string tmp(frame_data, frame_size);
std::cout << tmp << std::endl;
rapidjson::Document json;
THROW_WALLET_EXCEPTION_IF(json.Parse(frame_data, frame_size).HasParseError(), error::get_random_outs_error, "Couldn't JSON parse random outputs.");
for (rapidjson::SizeType i = 0; i < json["outputs"].Size(); i++) {
COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::outs_for_amount output;
output.amount = json["outputs"][i]["amount"].GetInt();
for (rapidjson::SizeType j = 0; j < json["outputs"][i]["outs"].Size(); j++) {
COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::out_entry entry;
entry.global_amount_index = json["outputs"][i]["outs"][j]["global_amount_index"].GetInt();
std::string out_key(json["outputs"][i]["outs"][j]["out_key"].GetString(), json["outputs"][i]["outs"][j]["out_key"].GetStringLength());
memcpy(entry.out_key.data, out_key.c_str(), 32);
output.outs.push_back(entry);
}
daemon_resp.outs.push_back(output);
}
std::vector<COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::outs_for_amount> scanty_outs; std::vector<COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::outs_for_amount> scanty_outs;
BOOST_FOREACH(COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::outs_for_amount& amount_outs, daemon_resp.outs) BOOST_FOREACH(COMMAND_RPC_GET_RANDOM_OUTPUTS_FOR_AMOUNTS::outs_for_amount& amount_outs, daemon_resp.outs)

6
src/wallet/wallet_errors.h
View File

@ -439,7 +439,7 @@ namespace tools
//---------------------------------------------------------------------------------------------------- //----------------------------------------------------------------------------------------------------
struct tx_rejected : public transfer_error struct tx_rejected : public transfer_error
{ {
explicit tx_rejected(std::string&& loc, const cryptonote::transaction& tx, const std::string& status) explicit tx_rejected(std::string&& loc, const cryptonote::transaction& tx, uint64_t status)
: transfer_error(std::move(loc), "transaction was rejected by daemon") : transfer_error(std::move(loc), "transaction was rejected by daemon")
, m_tx(tx) , m_tx(tx)
, m_status(status) , m_status(status)
@ -447,7 +447,7 @@ namespace tools
} }
const cryptonote::transaction& tx() const { return m_tx; } const cryptonote::transaction& tx() const { return m_tx; }
const std::string& status() const { return m_status; } uint64_t status() const { return m_status; }
std::string to_string() const std::string to_string() const
{ {
@ -460,7 +460,7 @@ namespace tools
private: private:
cryptonote::transaction m_tx; cryptonote::transaction m_tx;
std::string m_status; uint64_t m_status;
}; };
//---------------------------------------------------------------------------------------------------- //----------------------------------------------------------------------------------------------------
struct tx_sum_overflow : public transfer_error struct tx_sum_overflow : public transfer_error