static int
pcre_compile_do(struct sol_flow_node *node,
    const char *regexp,
    pcre **compiled_re,
    pcre_extra **p_extra,
    int *sub_match_count)
{
    int r;
    int pcre_error_offset;
    const char *pcre_error_str = NULL;

    SOL_NULL_CHECK(compiled_re, -EINVAL);
    SOL_NULL_CHECK(p_extra, -EINVAL);
    SOL_NULL_CHECK(sub_match_count, -EINVAL);

    *compiled_re = NULL;
    *p_extra = NULL;
    *sub_match_count = 0;

    *compiled_re = pcre_compile(regexp, PCRE_UTF8,
        &pcre_error_str, &pcre_error_offset, NULL);
    if (!*compiled_re) {
        sol_flow_send_error_packet(node, EINVAL,
            "Could not compile '%s': %s", regexp, pcre_error_str);
        return -EINVAL;
    }

    r = pcre_fullinfo(*compiled_re, NULL, PCRE_INFO_CAPTURECOUNT,
        sub_match_count);
    if (r < 0) {
        sol_flow_send_error_packet(node, EINVAL, "Could not"
            "extract info from compiled regular expression %s", regexp);
        pcre_free(compiled_re);
        return -EINVAL;
    }

    //A null pcre_extra is fine (no optimization possible), errors are
    //reported on the last argument
    *p_extra = pcre_study(*compiled_re, 0, &pcre_error_str);
    if (pcre_error_str != NULL) {
        sol_flow_send_error_packet(node, EINVAL,
            "Error optimizing '%s': %s", regexp, pcre_error_str);
        pcre_free(compiled_re);
        return -EINVAL;
    }

    return 0;
}

static void
_lsm303_send_output_packets(struct sol_flow_node *node, struct magnetometer_lsm303_data *mdata)
{
    struct sol_direction_vector val =
    {
        .min = -mdata->scale,
        .max = -mdata->scale,
        .x = mdata->reading[0],
        .y = mdata->reading[1],
        .z = mdata->reading[2]
    };

    sol_flow_send_direction_vector_packet(node,
        SOL_FLOW_NODE_TYPE_MAGNETOMETER_LSM303__OUT__OUT, &val);
}

static int
magnetometer_lsm303_tick(struct sol_flow_node *node, void *data, uint16_t port, uint16_t conn_id, const struct sol_flow_packet *packet)
{
    struct magnetometer_lsm303_data *mdata = data;
    int8_t buffer[LSM303_MAG_BYTES_NUMBER];
    int r;

    if (!sol_i2c_set_slave_address(mdata->i2c, mdata->slave)) {
        const char errmsg[] = "Failed to set slave at address 0x%02x";
        SOL_WRN(errmsg, mdata->slave);
        sol_flow_send_error_packet(node, EIO, errmsg, mdata->slave);
        return -EIO;
    }

    r = sol_i2c_read_register(mdata->i2c, LSM303_ACCEL_REG_OUT_X_H_M,
        (uint8_t *)buffer, sizeof(buffer));
    if (r <= 0) {
        const char errmsg[] = "Failed to read LSM303 magnetometer samples";
        SOL_WRN(errmsg);
        sol_flow_send_error_packet(node, EIO, errmsg);
        return -EIO;
    }

    /* Get X, Z and Y. That's why reading[] is indexed 0, 2 and 1. */
    mdata->reading[0] = ((buffer[0] << 8) | buffer[1]) / mdata->gain_xy;
    mdata->reading[2] = ((buffer[2] << 8) | buffer[3]) / mdata->gain_z;
    mdata->reading[1] = ((buffer[4] << 8) | buffer[5]) / mdata->gain_xy;

    _lsm303_send_output_packets(node, mdata);

    return 0;
}

int
sequence_process(
    struct sol_flow_node *node,
    void *data,
    uint16_t port,
    uint16_t conn_id,
    const struct sol_flow_packet *packet)
{
    struct boolean_validator_data *mdata = data;
    char input;
    bool val, match;
    int r;

    if (mdata->done) {
        sol_flow_send_error_packet(node, ECANCELED,
            "Input stream already deviated from expected data, ignoring packets.");
        return 0;
    }

    r = sol_flow_packet_get_boolean(packet, &val);
    SOL_INT_CHECK(r, < 0, r);

    input = val ? 'T' : 'F';

    match = input == *mdata->it;

    mdata->it++;
    if (*mdata->it == '\0' || !match) {
        sol_flow_send_boolean_packet(node, SOL_FLOW_NODE_TYPE_TEST_BOOLEAN_VALIDATOR__OUT__OUT,
            match);
        mdata->done = true;
    }
    return 0;
}

int
float_validator_process(
    struct sol_flow_node *node,
    void *data,
    uint16_t port,
    uint16_t conn_id,
    const struct sol_flow_packet *packet)
{
    struct float_validator_data *mdata = data;
    struct sol_drange input;
    double *op;
    bool match;

    if (mdata->done) {
        sol_flow_send_error_packet(node, ECANCELED,
            "Input stream already deviated from expected data, ignoring packets.");
        return 0;
    }

    sol_flow_packet_get_drange(packet, &input);
    op = sol_vector_get(&mdata->values, mdata->next_index);
    match = sol_drange_val_equal(input.val, *op);
    mdata->next_index++;

    if (mdata->next_index == mdata->values.len || !match) {
        sol_flow_send_boolean_packet(node, SOL_FLOW_NODE_TYPE_TEST_FLOAT_VALIDATOR__OUT__OUT,
            match);
        mdata->done = true;
    }
    return 0;
}

int
blob_validator_process(
    struct sol_flow_node *node,
    void *data,
    uint16_t port,
    uint16_t conn_id,
    const struct sol_flow_packet *packet)
{
    struct blob_validator_data *mdata = data;
    struct sol_blob *val;
    bool match;
    int r;

    if (mdata->done) {
        sol_flow_send_error_packet(node, ECANCELED,
            "Input stream already deviated from expected data, ignoring packets.");
        return 0;
    }

    r = sol_flow_packet_get_blob(packet, &val);
    SOL_INT_CHECK(r, < 0, r);
    match = (mdata->expected.size == val->size) && memcmp(mdata->expected.mem, val->mem, val->size) == 0;

    sol_flow_send_bool_packet(node, SOL_FLOW_NODE_TYPE_TEST_BOOLEAN_VALIDATOR__OUT__OUT, match);
    mdata->done = true;

    return 0;
}

static bool
timer_tick(void *data)
{
    struct sol_flow_node *node = data;
    struct boolean_generator_data *mdata = sol_flow_node_get_private_data(node);
    bool out_packet;

    if (*mdata->it == 'T') {
        out_packet = true;
    } else if (*mdata->it == 'F') {
        out_packet = false;
    } else {
        mdata->timer = NULL;
        sol_flow_send_error_packet(node, ECANCELED,
            "Unknown sample: %c. Option 'sequence' must be composed by 'T' and/or 'F' chars.",
            *mdata->it);
        return false;
    }

    sol_flow_send_boolean_packet(node, SOL_FLOW_NODE_TYPE_TEST_BOOLEAN_GENERATOR__OUT__OUT,
        out_packet);

    mdata->it++;
    return *mdata->it != '\0';
}

static int
value_set(struct sol_flow_node *node, void *data, uint16_t port, uint16_t conn_id, const struct sol_flow_packet *packet)
{
    int32_t value;
    int r;
    struct segments_ctl_data *mdata = data;

    if (mdata->needs_clear) {
        _clear(node);
        mdata->needs_clear = false;
    }

#define RANGE_MIN (0)
#define RANGE_MAX (15)

    r = sol_flow_packet_get_irange_value(packet, &value);
    SOL_INT_CHECK(r, < 0, r);

    if ((value < RANGE_MIN) || (value > RANGE_MAX)) {
        sol_flow_send_error_packet(node, ERANGE,
            "Range invalid, it should be between %d and %d but was %" PRId32,
            RANGE_MIN, RANGE_MAX, value);
        return 0;
    }
    _write_byte(node, font[value]);

    return 0;

#undef RANGE_MAX
#undef RANGE_MIN
}

static int
operator_process(struct sol_flow_node *node, void *data, uint16_t port, uint16_t conn_id, const struct sol_flow_packet *packet)
{
    struct drange_arithmetic_data *mdata = data;
    struct sol_drange value;
    int r;

    r = sol_flow_packet_get_drange(packet, &value);

    if (r < 0) {
        sol_flow_send_error_packet(node, -r, sol_util_strerrora(-r));
        return r;
    }

    if (port == 0) {
        mdata->var0 = value;
        mdata->var0_initialized = true;
    } else {
        mdata->var1 = value;
        mdata->var1_initialized = true;
    }

    /* wait until have both variables */
    if (!(mdata->var0_initialized && mdata->var1_initialized))
        return 0;

    r = mdata->func(&mdata->var0, &mdata->var1, &value);
    SOL_INT_CHECK(r, < 0, r);

    return sol_flow_send_drange_packet(node, mdata->port, &value);
}

static void
read_cb(void *cb_data, struct sol_aio *aio, int32_t ret)
{
    struct sol_irange i;
    struct aio_data *mdata = cb_data;

    SOL_NULL_CHECK(mdata);

    i.val = ret;
    if (i.val < 0) {
        sol_flow_send_error_packet(mdata->node, EINVAL,
            "AIO (%s): Failed on read operation: %s.", mdata->pin,
            sol_util_strerrora(i.val));
        return;
    }

    if (mdata->is_first || i.val != mdata->last_value) {
        mdata->is_first = false;
        mdata->last_value = i.val;
        i.max = mdata->mask;
        i.min = 0;
        i.step = 1;
        sol_flow_send_irange_packet(mdata->node,
            SOL_FLOW_NODE_TYPE_AIO_READER__OUT__OUT, &i);
    }

    mdata->pending = NULL;
}

static void
check_cb(void *data, int status, const struct sol_update_info *response)
{
    struct sol_flow_node *node = data;
    struct update_data *mdata = sol_flow_node_get_private_data(node);

    if (status < 0) {
        sol_flow_send_error_packet(node, -status,
            "Error while checking for updates: %s", sol_util_strerrora(-status));
        goto end;
    }

#ifndef SOL_NO_API_VERSION
    if (SOL_UNLIKELY(response->api_version != SOL_UPDATE_INFO_API_VERSION)) {
        SOL_WRN("Update info config version '%u' is unexpected, expected '%u'",
            response->api_version, SOL_UPDATE_INFO_API_VERSION);
        return;
    }
#endif

    sol_flow_send_string_packet(node,
        SOL_FLOW_NODE_TYPE_UPDATE_CHECK__OUT__VERSION, response->version);
    sol_flow_send_irange_value_packet(node,
        SOL_FLOW_NODE_TYPE_UPDATE_CHECK__OUT__SIZE, response->size);
    sol_flow_send_bool_packet(node,
        SOL_FLOW_NODE_TYPE_UPDATE_CHECK__OUT__NEED_UPDATE, response->need_update);

end:
    mdata->handle = NULL;
}

static int
common_get_progress(struct sol_flow_node *node, void *data, uint16_t port, uint16_t conn_id, const struct sol_flow_packet *packet)
{
    struct update_data *mdata = data;
    struct update_node_type *type;
    struct sol_irange irange = {
        .step = 1,
        .min = 0,
        .max = 100
    };

    type = (struct update_node_type *)sol_flow_node_get_type(node);

    if (mdata->handle) {
        irange.val = sol_update_get_progress(mdata->handle);
        if (irange.val >= 0 && irange.val <= 100)
            sol_flow_send_irange_packet(node, type->progress_port, &irange);
        else
            sol_flow_send_error_packet(node, EINVAL,
                "Could not get progress of task");
    } else {
        SOL_DBG("No current operation in process, ignoring request to get progress");
    }

    return 0;
}

int
byte_validator_process(
    struct sol_flow_node *node,
    void *data,
    uint16_t port,
    uint16_t conn_id,
    const struct sol_flow_packet *packet)
{
    struct byte_validator_data *mdata = data;
    unsigned char *op;
    int r;
    unsigned char val;
    bool match;

    if (mdata->done) {
        sol_flow_send_error_packet(node, ECANCELED,
            "Input stream already deviated from expected data, ignoring packets.");
        return 0;
    }
    r = sol_flow_packet_get_byte(packet, &val);
    SOL_INT_CHECK(r, < 0, r);
    op = sol_vector_get(&mdata->values, mdata->next_index);
    match = val == *op;
    mdata->next_index++;

    if (mdata->next_index == mdata->values.len || !match) {
        sol_flow_send_boolean_packet(node,
            SOL_FLOW_NODE_TYPE_TEST_INT_VALIDATOR__OUT__OUT, match);
        mdata->done = true;
    }
    return 0;
}

static void
thingspeak_execute_poll_finished(void *data,
    const struct sol_http_client_connection *connection,
    struct sol_http_response *response)
{
    struct thingspeak_execute_data *mdata = data;
    char *body;

    RESPONSE_CHECK_API(response, mdata);

    if (!response->content.used) {
        /* Empty response means no talkback commands */
        return;
    }

    if (response->response_code != SOL_HTTP_STATUS_OK) {
        sol_flow_send_error_packet(mdata->node, EINVAL,
            "Thingspeak returned an unknown response code: %d",
            response->response_code);
        return;
    }

    body = strndup(response->content.data, response->content.used);
    SOL_NULL_CHECK(body);

    sol_flow_send_string_packet(mdata->node,
        SOL_FLOW_NODE_TYPE_THINGSPEAK_TALKBACK_EXECUTE__OUT__OUT, body);
    free(body);
}

SOL_API int
sol_flow_send_error_packet_errno(struct sol_flow_node *src, int code)
{
    if (code < 0)
        code = -code;
    return sol_flow_send_error_packet(src, code, "%s (errno %d)", sol_util_strerrora(code), code);
}

static int
platform_service_stop_process(struct sol_flow_node *node, void *data, uint16_t port, uint16_t conn_id, const struct sol_flow_packet *packet)
{
    struct platform_service_data *mdata = data;

    if (!mdata->service_name) {
        return sol_flow_send_error_packet(mdata->node, EINVAL,
            "Service name not set");
    }

    return sol_platform_stop_service(mdata->service_name);
}

/*
 * This function takes mdata->to_regexp and copies it into a struct
 * sol_buffer, with all the number ref entries already replaced by the
 * due text. It assumes that @a buf is already initialized.
 */
static int
get_unescaped_regexp_replacement_str(struct string_regexp_replace_data *mdata,
    struct sol_buffer *buf,
    int *match_vector,
    size_t match_cnt)
{
    int r;
    struct sol_str_slice slice;
    char *ptr = mdata->to_regexp;

    SOL_NULL_CHECK(buf, -EINVAL);

    /* \g<number> or \<number> syntax. Go on as far as numbers are in
     * [0-9] range. Leading zeros are permitted.
     */
    while (*ptr) {
        if (*ptr == '\\') {
            int n;
            uint64_t grp_num = 0;

            ptr++;

            if (*ptr == 'g')
                ptr++;

            for (; (n = *ptr - '0') >= 0 && n <= 9; ptr++) {
                if (!grp_num && !n) {
                    continue;
                }

                r = sol_util_uint64_mul(grp_num, 10, &grp_num);
                SOL_INT_CHECK_GOTO(r, < 0, err);

                r = sol_util_uint64_add(grp_num, n, &grp_num);
                SOL_INT_CHECK_GOTO(r, < 0, err);
            }

            if (!grp_num || grp_num > match_cnt) {
                sol_flow_send_error_packet(mdata->node, EINVAL,
                    "Could not process '%s' pattern's reference "
                    "to non-existent subpattern on '%s'",
                    mdata->to_regexp, mdata->regexp);
                r = -EINVAL;
                goto err;
            }

            slice = SOL_STR_SLICE_STR(mdata->orig_string
                + match_vector[grp_num * 2],
                match_vector[grp_num * 2 + 1] - match_vector[grp_num * 2]);
            r = sol_buffer_append_slice(buf, slice);
            SOL_INT_CHECK_GOTO(r, < 0, err);
        } else {

static void
thingspeak_add_request_finished(void *data,
    const struct sol_http_client_connection *connection,
    struct sol_http_response *response)
{
    struct thingspeak_add_data *mdata = data;

    RESPONSE_CHECK_API(response, mdata);

    if (!response->content.used) {
        sol_flow_send_error_packet(mdata->node, EINVAL,
            "Command ID not received back from Thingspeak");
        return;
    }

    if (response->response_code != SOL_HTTP_STATUS_OK) {
        sol_flow_send_error_packet(mdata->node, EINVAL,
            "Thingspeak returned an unknown response code: %d",
            response->response_code);
        return;
    }
}

static void
thingspeak_channel_update_finished(void *data,
    const struct sol_http_client_connection *connection,
    struct sol_http_response *response)
{
    struct thingspeak_channel_update_data *mdata = data;

    RESPONSE_CHECK_API(response, mdata);

    if (!strncmp(response->content.data, "0", response->content.used)) {
        sol_flow_send_error_packet(mdata->node, EINVAL,
            "Could not update Thingspeak channel");
    }
}

static bool
_on_reader_timeout(void *data)
{
    struct aio_data *mdata = data;

    mdata->pending = sol_aio_get_value(mdata->aio, read_cb, mdata);
    if (!mdata->pending && errno != EBUSY) {
        sol_flow_send_error_packet(mdata->node, errno,
            "AIO (%s): Failed to issue read operation.", mdata->pin);
        return false;
    }

    return true;
}

static void
install_cb(void *data, int status)
{
    struct sol_flow_node *node = data;
    struct update_data *mdata = sol_flow_node_get_private_data(node);

    if (status < 0) {
        sol_flow_send_error_packet(node, -status,
            "Error while installing update: %s", sol_util_strerrora(-status));
    }

    sol_flow_send_bool_packet(node,
        SOL_FLOW_NODE_TYPE_UPDATE_INSTALL__OUT__SUCCESS, status == 0);

    mdata->handle = NULL;
}