doveadm/dsync/dsync-ibc-pipe.c

9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek/* Copyright (c) 2013-2018 Dovecot authors, see the included COPYING file */
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek#include "lib.h"
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek#include "array.h"
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek#include "istream.h"
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek#include "dsync-mail.h"
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek#include "dsync-mailbox.h"
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek#include "dsync-mailbox-state.h"
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek#include "dsync-mailbox-tree.h"
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek#include "dsync-ibc-private.h"
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekenum item_type {
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ITEM_END_OF_LIST,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ITEM_HANDSHAKE,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ITEM_MAILBOX_STATE,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ITEM_MAILBOX_TREE_NODE,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ITEM_MAILBOX_DELETE,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ITEM_MAILBOX,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ITEM_MAILBOX_ATTRIBUTE,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ITEM_MAIL_CHANGE,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ITEM_MAIL_REQUEST,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ITEM_MAIL,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ITEM_FINISH
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek};
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekstruct item {
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    enum item_type type;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    pool_t pool;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    union {
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        struct dsync_ibc_settings set;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        struct dsync_mailbox_state state;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        struct dsync_mailbox_node node;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        guid_128_t mailbox_guid;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        struct dsync_mailbox dsync_box;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        struct dsync_mailbox_attribute attr;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        struct dsync_mail_change change;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        struct dsync_mail_request request;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        struct dsync_mail mail;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        struct {
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek            const struct dsync_mailbox_delete *deletes;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek            unsigned int count;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek            char hierarchy_sep;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        } mailbox_delete;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        struct {
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek            const char *error;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek            enum mail_error mail_error;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek            bool require_full_resync;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        } finish;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    } u;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek};
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekstruct dsync_ibc_pipe {
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    struct dsync_ibc ibc;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ARRAY(pool_t) pools;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ARRAY(struct item) item_queue;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    struct dsync_ibc_pipe *remote;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    pool_t pop_pool;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    struct item pop_item;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek};
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekstatic pool_t dsync_ibc_pipe_get_pool(struct dsync_ibc_pipe *pipe)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek{
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    pool_t *pools, ret;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    unsigned int count;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    pools = array_get_modifiable(&pipe->pools, &count);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    if (count == 0)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        return pool_alloconly_create(MEMPOOL_GROWING"pipe item pool", 1024);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    ret = pools[count-1];
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    array_delete(&pipe->pools, count-1, 1);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    p_clear(ret);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    return ret;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek}
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekstatic struct item * ATTR_NOWARN_UNUSED_RESULT
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekdsync_ibc_pipe_push_item(struct dsync_ibc_pipe *pipe, enum item_type type)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek{
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    struct item *item;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item = array_append_space(&pipe->item_queue);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item->type = type;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    switch (type) {
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    case ITEM_END_OF_LIST:
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    case ITEM_MAILBOX_STATE:
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    case ITEM_MAILBOX_DELETE:
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        break;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    case ITEM_HANDSHAKE:
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    case ITEM_MAILBOX:
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    case ITEM_MAILBOX_TREE_NODE:
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    case ITEM_MAILBOX_ATTRIBUTE:
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    case ITEM_MAIL_CHANGE:
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    case ITEM_MAIL_REQUEST:
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    case ITEM_MAIL:
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    case ITEM_FINISH:
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        item->pool = dsync_ibc_pipe_get_pool(pipe);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        break;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    }
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    return item;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek}
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekstatic struct item *
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekdsync_ibc_pipe_pop_item(struct dsync_ibc_pipe *pipe, enum item_type type)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek{
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    struct item *item;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    if (array_count(&pipe->item_queue) == 0)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        return NULL;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item = array_idx_modifiable(&pipe->item_queue, 0);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    i_assert(item->type == type);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    pipe->pop_item = *item;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    array_delete(&pipe->item_queue, 0, 1);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item = NULL;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    pool_unref(&pipe->pop_pool);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    pipe->pop_pool = pipe->pop_item.pool;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    return &pipe->pop_item;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek}
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekstatic bool dsync_ibc_pipe_try_pop_eol(struct dsync_ibc_pipe *pipe)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek{
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    const struct item *item;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    if (array_count(&pipe->item_queue) == 0)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        return FALSE;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item = array_idx(&pipe->item_queue, 0);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    if (item->type != ITEM_END_OF_LIST)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        return FALSE;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    array_delete(&pipe->item_queue, 0, 1);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    return TRUE;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek}
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekstatic void dsync_ibc_pipe_deinit(struct dsync_ibc *ibc)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek{
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    struct item *item;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    pool_t *poolp;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    if (pipe->remote != NULL) {
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        i_assert(pipe->remote->remote == pipe);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        pipe->remote->remote = NULL;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    }
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    pool_unref(&pipe->pop_pool);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    array_foreach_modifiable(&pipe->item_queue, item) {
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        pool_unref(&item->pool);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    }
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    array_foreach_modifiable(&pipe->pools, poolp)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        pool_unref(poolp);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    array_free(&pipe->pools);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    array_free(&pipe->item_queue);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    i_free(pipe);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek}
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekstatic void
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekdsync_ibc_pipe_send_handshake(struct dsync_ibc *ibc,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek                  const struct dsync_ibc_settings *set)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek{
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    struct item *item;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item = dsync_ibc_pipe_push_item(pipe->remote, ITEM_HANDSHAKE);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item->u.set = *set;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item->u.set.sync_ns_prefixes =
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        p_strdup(item->pool, set->sync_ns_prefixes);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item->u.set.sync_box = p_strdup(item->pool, set->sync_box);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item->u.set.virtual_all_box = p_strdup(item->pool, set->virtual_all_box);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item->u.set.exclude_mailboxes = set->exclude_mailboxes == NULL ? NULL :
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        p_strarray_dup(item->pool, set->exclude_mailboxes);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    memcpy(item->u.set.sync_box_guid, set->sync_box_guid,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek           sizeof(item->u.set.sync_box_guid));
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item->u.set.sync_since_timestamp = set->sync_since_timestamp;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item->u.set.sync_until_timestamp = set->sync_until_timestamp;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item->u.set.sync_max_size = set->sync_max_size;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item->u.set.sync_flags = p_strdup(item->pool, set->sync_flags);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek}
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekstatic enum dsync_ibc_recv_ret
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozekdsync_ibc_pipe_recv_handshake(struct dsync_ibc *ibc,
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek                  const struct dsync_ibc_settings **set_r)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek{
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    struct item *item;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    item = dsync_ibc_pipe_pop_item(pipe, ITEM_HANDSHAKE);
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek    if (item == NULL)
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek        return DSYNC_IBC_RECV_RET_TRYAGAIN;
9a9b5e115b079751422be22fd252c0b283611c62Jakub Hrozek
    *set_r = &item->u.set;
    return DSYNC_IBC_RECV_RET_OK;
}

static bool dsync_ibc_pipe_is_send_queue_full(struct dsync_ibc *ibc)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;

    return array_count(&pipe->remote->item_queue) > 0;
}

static bool dsync_ibc_pipe_has_pending_data(struct dsync_ibc *ibc)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;

    return array_count(&pipe->item_queue) > 0;
}

static void
dsync_ibc_pipe_send_end_of_list(struct dsync_ibc *ibc,
                enum dsync_ibc_eol_type type ATTR_UNUSED)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;

    dsync_ibc_pipe_push_item(pipe->remote, ITEM_END_OF_LIST);
}

static void
dsync_ibc_pipe_send_mailbox_state(struct dsync_ibc *ibc,
                  const struct dsync_mailbox_state *state)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    item = dsync_ibc_pipe_push_item(pipe->remote, ITEM_MAILBOX_STATE);
    item->u.state = *state;
}

static enum dsync_ibc_recv_ret
dsync_ibc_pipe_recv_mailbox_state(struct dsync_ibc *ibc,
                  struct dsync_mailbox_state *state_r)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    if (dsync_ibc_pipe_try_pop_eol(pipe))
        return DSYNC_IBC_RECV_RET_FINISHED;

    item = dsync_ibc_pipe_pop_item(pipe, ITEM_MAILBOX_STATE);
    if (item == NULL)
        return DSYNC_IBC_RECV_RET_TRYAGAIN;

    *state_r = item->u.state;
    return DSYNC_IBC_RECV_RET_OK;
}

static void
dsync_ibc_pipe_send_mailbox_tree_node(struct dsync_ibc *ibc,
                      const char *const *name,
                      const struct dsync_mailbox_node *node)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    item = dsync_ibc_pipe_push_item(pipe->remote, ITEM_MAILBOX_TREE_NODE);

    /* a little bit kludgy way to send it */
    item->u.node.name = (void *)p_strarray_dup(item->pool, name);
    dsync_mailbox_node_copy_data(&item->u.node, node);
}

static enum dsync_ibc_recv_ret
dsync_ibc_pipe_recv_mailbox_tree_node(struct dsync_ibc *ibc,
                      const char *const **name_r,
                      const struct dsync_mailbox_node **node_r)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    if (dsync_ibc_pipe_try_pop_eol(pipe))
        return DSYNC_IBC_RECV_RET_FINISHED;

    item = dsync_ibc_pipe_pop_item(pipe, ITEM_MAILBOX_TREE_NODE);
    if (item == NULL)
        return DSYNC_IBC_RECV_RET_TRYAGAIN;

    *name_r = (void *)item->u.node.name;
    item->u.node.name = NULL;

    *node_r = &item->u.node;
    return DSYNC_IBC_RECV_RET_OK;
}

static void
dsync_ibc_pipe_send_mailbox_deletes(struct dsync_ibc *ibc,
                    const struct dsync_mailbox_delete *deletes,
                    unsigned int count, char hierarchy_sep)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    item = dsync_ibc_pipe_push_item(pipe->remote, ITEM_MAILBOX_DELETE);

    /* we'll assume that the deletes are permanent. this works for now.. */
    /* a little bit kludgy way to send it */
    item->u.mailbox_delete.deletes = deletes;
    item->u.mailbox_delete.count = count;
    item->u.mailbox_delete.hierarchy_sep = hierarchy_sep;
}

static enum dsync_ibc_recv_ret
dsync_ibc_pipe_recv_mailbox_deletes(struct dsync_ibc *ibc,
                    const struct dsync_mailbox_delete **deletes_r,
                    unsigned int *count_r,
                    char *hierarchy_sep_r)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    if (dsync_ibc_pipe_try_pop_eol(pipe))
        return DSYNC_IBC_RECV_RET_FINISHED;

    item = dsync_ibc_pipe_pop_item(pipe, ITEM_MAILBOX_DELETE);
    if (item == NULL)
        return DSYNC_IBC_RECV_RET_TRYAGAIN;

    *deletes_r = item->u.mailbox_delete.deletes;
    *count_r = item->u.mailbox_delete.count;
    *hierarchy_sep_r = item->u.mailbox_delete.hierarchy_sep;
    return DSYNC_IBC_RECV_RET_OK;
}

static void
dsync_ibc_pipe_send_mailbox(struct dsync_ibc *ibc,
                const struct dsync_mailbox *dsync_box)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;
    const struct mailbox_cache_field *cf;
    struct mailbox_cache_field *ncf;

    item = dsync_ibc_pipe_push_item(pipe->remote, ITEM_MAILBOX);
    item->u.dsync_box = *dsync_box;
    p_array_init(&item->u.dsync_box.cache_fields, item->pool,
             array_count(&dsync_box->cache_fields));
    array_foreach(&dsync_box->cache_fields, cf) {
        ncf = array_append_space(&item->u.dsync_box.cache_fields);
        ncf->name = p_strdup(item->pool, cf->name);
        ncf->decision = cf->decision;
        ncf->last_used = cf->last_used;
    }
}

static enum dsync_ibc_recv_ret
dsync_ibc_pipe_recv_mailbox(struct dsync_ibc *ibc,
                const struct dsync_mailbox **dsync_box_r)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    if (dsync_ibc_pipe_try_pop_eol(pipe))
        return DSYNC_IBC_RECV_RET_FINISHED;

    item = dsync_ibc_pipe_pop_item(pipe, ITEM_MAILBOX);
    if (item == NULL)
        return DSYNC_IBC_RECV_RET_TRYAGAIN;

    *dsync_box_r = &item->u.dsync_box;
    return DSYNC_IBC_RECV_RET_OK;
}

static void
dsync_ibc_pipe_send_mailbox_attribute(struct dsync_ibc *ibc,
                      const struct dsync_mailbox_attribute *attr)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    item = dsync_ibc_pipe_push_item(pipe->remote, ITEM_MAILBOX_ATTRIBUTE);
    dsync_mailbox_attribute_dup(item->pool, attr, &item->u.attr);
}

static enum dsync_ibc_recv_ret
dsync_ibc_pipe_recv_mailbox_attribute(struct dsync_ibc *ibc,
                      const struct dsync_mailbox_attribute **attr_r)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    if (dsync_ibc_pipe_try_pop_eol(pipe))
        return DSYNC_IBC_RECV_RET_FINISHED;

    item = dsync_ibc_pipe_pop_item(pipe, ITEM_MAILBOX_ATTRIBUTE);
    if (item == NULL)
        return DSYNC_IBC_RECV_RET_TRYAGAIN;

    *attr_r = &item->u.attr;
    return DSYNC_IBC_RECV_RET_OK;
}

static void
dsync_ibc_pipe_send_change(struct dsync_ibc *ibc,
               const struct dsync_mail_change *change)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    item = dsync_ibc_pipe_push_item(pipe->remote, ITEM_MAIL_CHANGE);
    dsync_mail_change_dup(item->pool, change, &item->u.change);
}

static enum dsync_ibc_recv_ret
dsync_ibc_pipe_recv_change(struct dsync_ibc *ibc,
               const struct dsync_mail_change **change_r)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    if (dsync_ibc_pipe_try_pop_eol(pipe))
        return DSYNC_IBC_RECV_RET_FINISHED;

    item = dsync_ibc_pipe_pop_item(pipe, ITEM_MAIL_CHANGE);
    if (item == NULL)
        return DSYNC_IBC_RECV_RET_TRYAGAIN;

    *change_r = &item->u.change;
    return DSYNC_IBC_RECV_RET_OK;
}

static void
dsync_ibc_pipe_send_mail_request(struct dsync_ibc *ibc,
                 const struct dsync_mail_request *request)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    item = dsync_ibc_pipe_push_item(pipe->remote, ITEM_MAIL_REQUEST);
    item->u.request.guid = p_strdup(item->pool, request->guid);
    item->u.request.uid = request->uid;
}

static enum dsync_ibc_recv_ret
dsync_ibc_pipe_recv_mail_request(struct dsync_ibc *ibc,
                 const struct dsync_mail_request **request_r)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    if (dsync_ibc_pipe_try_pop_eol(pipe))
        return DSYNC_IBC_RECV_RET_FINISHED;

    item = dsync_ibc_pipe_pop_item(pipe, ITEM_MAIL_REQUEST);
    if (item == NULL)
        return DSYNC_IBC_RECV_RET_TRYAGAIN;

    *request_r = &item->u.request;
    return DSYNC_IBC_RECV_RET_OK;
}

static void
dsync_ibc_pipe_send_mail(struct dsync_ibc *ibc, const struct dsync_mail *mail)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    item = dsync_ibc_pipe_push_item(pipe->remote, ITEM_MAIL);
    item->u.mail.guid = p_strdup(item->pool, mail->guid);
    item->u.mail.uid = mail->uid;
    item->u.mail.pop3_uidl = p_strdup(item->pool, mail->pop3_uidl);
    item->u.mail.pop3_order = mail->pop3_order;
    item->u.mail.received_date = mail->received_date;
    if (mail->input != NULL) {
        item->u.mail.input = mail->input;
        i_stream_ref(mail->input);
    }
    item->u.mail.input_mail = mail->input_mail;
    item->u.mail.input_mail_uid = mail->input_mail_uid;
}

static enum dsync_ibc_recv_ret
dsync_ibc_pipe_recv_mail(struct dsync_ibc *ibc, struct dsync_mail **mail_r)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    if (dsync_ibc_pipe_try_pop_eol(pipe))
        return DSYNC_IBC_RECV_RET_FINISHED;

    item = dsync_ibc_pipe_pop_item(pipe, ITEM_MAIL);
    if (item == NULL)
        return DSYNC_IBC_RECV_RET_TRYAGAIN;

    *mail_r = &item->u.mail;
    return DSYNC_IBC_RECV_RET_OK;
}

static void
dsync_ibc_pipe_send_finish(struct dsync_ibc *ibc, const char *error,
               enum mail_error mail_error,
               bool require_full_resync)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    item = dsync_ibc_pipe_push_item(pipe->remote, ITEM_FINISH);
    item->u.finish.error = p_strdup(item->pool, error);
    item->u.finish.mail_error = mail_error;
    item->u.finish.require_full_resync = require_full_resync;
}

static enum dsync_ibc_recv_ret
dsync_ibc_pipe_recv_finish(struct dsync_ibc *ibc, const char **error_r,
               enum mail_error *mail_error_r,
               bool *require_full_resync_r)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;
    struct item *item;

    item = dsync_ibc_pipe_pop_item(pipe, ITEM_FINISH);
    if (item == NULL)
        return DSYNC_IBC_RECV_RET_TRYAGAIN;

    *error_r = item->u.finish.error;
    *mail_error_r = item->u.finish.mail_error;
    *require_full_resync_r = item->u.finish.require_full_resync;
    return DSYNC_IBC_RECV_RET_OK;
}

static void pipe_close_mail_streams(struct dsync_ibc_pipe *pipe)
{
    struct item *item;

    if (array_count(&pipe->item_queue) > 0) {
        item = array_idx_modifiable(&pipe->item_queue, 0);
        if (item->type == ITEM_MAIL &&
            item->u.mail.input != NULL)
            i_stream_unref(&item->u.mail.input);
    }
}

static void dsync_ibc_pipe_close_mail_streams(struct dsync_ibc *ibc)
{
    struct dsync_ibc_pipe *pipe = (struct dsync_ibc_pipe *)ibc;

    pipe_close_mail_streams(pipe);
    pipe_close_mail_streams(pipe->remote);
}

static const struct dsync_ibc_vfuncs dsync_ibc_pipe_vfuncs = {
    dsync_ibc_pipe_deinit,
    dsync_ibc_pipe_send_handshake,
    dsync_ibc_pipe_recv_handshake,
    dsync_ibc_pipe_send_end_of_list,
    dsync_ibc_pipe_send_mailbox_state,
    dsync_ibc_pipe_recv_mailbox_state,
    dsync_ibc_pipe_send_mailbox_tree_node,
    dsync_ibc_pipe_recv_mailbox_tree_node,
    dsync_ibc_pipe_send_mailbox_deletes,
    dsync_ibc_pipe_recv_mailbox_deletes,
    dsync_ibc_pipe_send_mailbox,
    dsync_ibc_pipe_recv_mailbox,
    dsync_ibc_pipe_send_mailbox_attribute,
    dsync_ibc_pipe_recv_mailbox_attribute,
    dsync_ibc_pipe_send_change,
    dsync_ibc_pipe_recv_change,
    dsync_ibc_pipe_send_mail_request,
    dsync_ibc_pipe_recv_mail_request,
    dsync_ibc_pipe_send_mail,
    dsync_ibc_pipe_recv_mail,
    dsync_ibc_pipe_send_finish,
    dsync_ibc_pipe_recv_finish,
    dsync_ibc_pipe_close_mail_streams,
    dsync_ibc_pipe_is_send_queue_full,
    dsync_ibc_pipe_has_pending_data
};

static struct dsync_ibc_pipe *
dsync_ibc_pipe_alloc(void)
{
    struct dsync_ibc_pipe *pipe;

    pipe = i_new(struct dsync_ibc_pipe, 1);
    pipe->ibc.v = dsync_ibc_pipe_vfuncs;
    i_array_init(&pipe->pools, 4);
    i_array_init(&pipe->item_queue, 4);
    return pipe;
}

void dsync_ibc_init_pipe(struct dsync_ibc **ibc1_r, struct dsync_ibc **ibc2_r)
{
    struct dsync_ibc_pipe *pipe1, *pipe2;

    pipe1 = dsync_ibc_pipe_alloc();
    pipe2 = dsync_ibc_pipe_alloc();
    pipe1->remote = pipe2;
    pipe2->remote = pipe1;
    *ibc1_r = &pipe1->ibc;
    *ibc2_r = &pipe2->ibc;
}