path: root/agent/tcf/framework/protocol.c

/*******************************************************************************
 * Copyright (c) 2007, 2010 Wind River Systems, Inc. and others.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 * and the Eclipse Distribution License is available at
 * http://www.eclipse.org/org/documents/edl-v10.php.
 * You may elect to redistribute this code under either of these licenses.
 *
 * Contributors:
 *     Wind River Systems - initial API and implementation
 *******************************************************************************/

/*
 * TCF communication protocol.
 * This module handles registration of command and event handlers.
 * It is called when new messages are received and will dispatch
 * messages to the appropriate handler. It has no knowledge of what transport
 * protocol is used and what services do.
 */

#include <tcf/config.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <assert.h>
#include <tcf/framework/protocol.h>
#include <tcf/framework/trace.h>
#include <tcf/framework/events.h>
#include <tcf/framework/events.h>
#include <tcf/framework/exceptions.h>
#include <tcf/framework/json.h>
#include <tcf/framework/myalloc.h>

static const char * LOCATOR = "Locator";

struct ServiceInfo {
    void * owner;
    char * name;
    struct ServiceInfo * next;
};

struct MessageHandlerInfo {
    Protocol * p;
    ServiceInfo * service;
    const char * name;
    ProtocolCommandHandler2 handler;
    void * client_data;
    struct MessageHandlerInfo * next;
};

typedef struct MessageHandlerInfo MessageHandlerInfo;

struct EventHandlerInfo {
    Channel * c;
    ServiceInfo * service;
    const char * name;
    ProtocolEventHandler2 handler;
    void * client_data;
    struct EventHandlerInfo * next;
};

typedef struct EventHandlerInfo EventHandlerInfo;

struct ReplyHandlerInfo {
    unsigned long tokenid;
    Channel * c;
    ReplyHandlerCB handler;
    void * client_data;
    struct ReplyHandlerInfo * next;
};

#define MESSAGE_HASH_SIZE (5 * MEM_USAGE_FACTOR - 1)
#define EVENT_HASH_SIZE   (4 * MEM_USAGE_FACTOR - 1)
#define REPLY_HASH_SIZE   (4 * MEM_USAGE_FACTOR - 1)

static MessageHandlerInfo * message_handlers[MESSAGE_HASH_SIZE];
static EventHandlerInfo * event_handlers[EVENT_HASH_SIZE];
static ReplyHandlerInfo * reply_handlers[REPLY_HASH_SIZE];
static ServiceInfo * services;
static int ini_done = 0;
static int proto_cnt = 0;
static char * agent_id = NULL;

struct Protocol {
    int id;
    int lock_cnt;           /* Lock count, cannot delete when > 0 */
    unsigned long tokenid;
    ProtocolMessageHandler2 default_handler;
    void * client_data;
};

static void read_stringz(InputStream * inp, char * str, size_t size) {
    unsigned len = 0;
    for (;;) {
        int ch = read_stream(inp);
        if (ch == 0) break;
        if (ch < 0) {
            trace(LOG_ALWAYS, "Unexpected end of message");
            exception(ERR_PROTOCOL);
        }
        if (len < size - 1) str[len++] = (char)ch;
    }
    str[len] = 0;
}

ServiceInfo * protocol_get_service(void * owner, const char * name) {
    ServiceInfo * s = services;

    while (s != NULL && (s->owner != owner || strcmp(s->name, name) != 0)) s = s->next;
    if (s == NULL) {
        assert(strcmp(name, "ZeroCopy") != 0);
        s = (ServiceInfo *)loc_alloc(sizeof(ServiceInfo));
        s->owner = owner;
        s->name = loc_strdup(name);
        s->next = services;
        services = s;
    }
    return s;
}

static void free_services(void * owner) {
    ServiceInfo ** sp = &services;
    ServiceInfo * s;

    while ((s = *sp) != NULL) {
        if (s->owner == owner) {
            *sp = s->next;
            loc_free(s->name);
            loc_free(s);
        }
        else {
            sp = &s->next;
        }
    }
}

static unsigned message_hash(Protocol * p, const char * service, const char * name) {
    int i;
    unsigned h = (unsigned)(uintptr_t)p >> 4;
    for (i = 0; service[i]; i++) h += service[i];
    for (i = 0; name[i]; i++) h += name[i];
    h = h + h / MESSAGE_HASH_SIZE;
    return h % MESSAGE_HASH_SIZE;
}

static MessageHandlerInfo * find_message_handler(Protocol * p, char * service, char * name) {
    MessageHandlerInfo * mh = message_handlers[message_hash(p, service, name)];
    while (mh != NULL) {
        if (mh->p == p && !strcmp(mh->service->name, service) && !strcmp(mh->name, name)) return mh;
        mh = mh->next;
    }
    return NULL;
}

static unsigned event_hash(Channel * c, const char * service, const char * name) {
    int i;
    unsigned h = (unsigned)(uintptr_t)c >> 4;
    for (i = 0; service[i]; i++) h += service[i];
    for (i = 0; name[i]; i++) h += name[i];
    h = h + h / EVENT_HASH_SIZE;
    return h % EVENT_HASH_SIZE;
}

static EventHandlerInfo * find_event_handler(Channel * c, char * service, char * name) {
    EventHandlerInfo * mh = event_handlers[event_hash(c, service, name)];
    while (mh != NULL) {
        if (mh->c == c && !strcmp(mh->service->name, service) && !strcmp(mh->name, name)) return mh;
        mh = mh->next;
    }
    return NULL;
}

#define reply_hash(c, tokenid) ((((unsigned)(uintptr_t)(c) >> 4) + (unsigned)(tokenid)) % REPLY_HASH_SIZE)

static ReplyHandlerInfo * find_reply_handler(Channel * c, unsigned long tokenid, int take) {
    ReplyHandlerInfo ** rhp = &reply_handlers[reply_hash(c, tokenid)];
    ReplyHandlerInfo * rh;
    while ((rh = *rhp) != NULL) {
        if (rh->c == c && rh->tokenid == tokenid) {
            if (take) *rhp = rh->next;
            return rh;
        }
        rhp = &rh->next;
    }
    return NULL;
}

static void skip_until_EOM(Channel * c) {
    for (;;) {
        int ch = read_stream(&c->inp);
        if (ch == MARKER_EOM) return;
        if (ch == MARKER_EOS) return;
    }
}

static void event_locator_hello(Channel * c);

void handle_protocol_message(Channel * c) {
    char type[8];
    char token[256];
    char service[256];
    char name[256];
    char * args[4];
    int error = 0;
    Protocol * p = c->protocol;

    assert(is_dispatch_thread());

    read_stringz(&c->inp, type, sizeof(type));
    if (strlen(type) != 1) {
        trace(LOG_ALWAYS, "Invalid TCF message: %s ...", type);
        error = ERR_PROTOCOL;
    }
    else if (type[0] == 'C') {
        Trap trap;
        read_stringz(&c->inp, token, sizeof(token));
        read_stringz(&c->inp, service, sizeof(service));
        read_stringz(&c->inp, name, sizeof(name));
        trace(LOG_PROTOCOL, "Peer %s: Command: C %s %s %s ...", c->peer_name, token, service, name);
        if (c->state != ChannelStateConnected) {
            trace(LOG_PROTOCOL, "Wrong channel state for commands");
            skip_until_EOM(c);
            write_stringz(&c->out, "N");
            write_stringz(&c->out, token);
            write_stream(&c->out, MARKER_EOM);
        }
        else if (set_trap(&trap)) {
            MessageHandlerInfo * mh = find_message_handler(p, service, name);
            if (mh != NULL) {
                mh->handler(token, c, mh->client_data);
            }
            else if (p->default_handler != NULL) {
                args[0] = type;
                args[1] = token;
                args[2] = service;
                args[3] = name;
                p->default_handler(c, args, 4, p->client_data);
            }
            else {
                trace(LOG_PROTOCOL, "Command is not recognized: %s %s ...", service, name);
                skip_until_EOM(c);
                write_stringz(&c->out, "N");
                write_stringz(&c->out, token);
                write_stream(&c->out, MARKER_EOM);
            }
            clear_trap(&trap);
        }
        else {
            trace(LOG_ALWAYS, "Exception handling command %s.%s: %d %s",
                service, name, trap.error, errno_to_str(trap.error));
            error = trap.error;
        }
    }
    else if (type[0] == 'R' || type[0] == 'P' || type[0] == 'N') {
        Trap trap;
        read_stringz(&c->inp, token, sizeof(token));
        trace(LOG_PROTOCOL, "Peer %s: Reply: %c %s ...", c->peer_name, type[0], token);
        if (set_trap(&trap)) {
            ReplyHandlerInfo * rh = NULL;
            char * endptr = NULL;
            unsigned long tokenid;
            errno = 0;
            tokenid = strtoul(token, &endptr, 10);
            if (errno != 0 || *endptr != '\0' ||
               (rh = find_reply_handler(c, tokenid, type[0] != 'P')) == NULL) {
                if (p->default_handler != NULL) {
                    args[0] = type;
                    args[1] = token;
                    p->default_handler(c, args, 2, p->client_data);
                }
                else {
                    trace(LOG_ALWAYS, "Reply with unexpected token: %s", token);
                    exception(ERR_PROTOCOL);
                }
            }
            else {
                int n = 0;
                if (type[0] == 'N') {
                    skip_until_EOM(c);
                    n = ERR_INV_COMMAND;
                }
                rh->handler(c, rh->client_data, n);
                if (type[0] != 'P') loc_free(rh);
            }
            clear_trap(&trap);
        }
        else {
            trace(LOG_ALWAYS, "Exception handling reply %s: %d %s",
                  token, trap.error, errno_to_str(trap.error));
            error = trap.error;
        }
    }
    else if (type[0] == 'E') {
        Trap trap;
        read_stringz(&c->inp, service, sizeof(service));
        read_stringz(&c->inp, name, sizeof(name));
        trace(LOG_PROTOCOL, "Peer %s: Event: E %s %s ...", c->peer_name, service, name);
        if (set_trap(&trap)) {
            if ((c->state == ChannelStateStarted || c->state == ChannelStateHelloSent) &&
                strcmp(service, LOCATOR) == 0 && strcmp(name, "Hello") == 0) {
                event_locator_hello(c);
            }
            else {
                EventHandlerInfo * eh = find_event_handler(c, service, name);
                if (eh != NULL) {
                    eh->handler(c, eh->client_data);
                }
                else if (p->default_handler != NULL) {
                    args[0] = type;
                    args[1] = service;
                    args[2] = name;
                    p->default_handler(c, args, 3, p->client_data);
                }
                else {
                    skip_until_EOM(c);
                }
            }
            clear_trap(&trap);
        }
        else {
            trace(LOG_ALWAYS, "Exception handling event %s.%s: %d %s",
                service, name, trap.error, errno_to_str(trap.error));
            error = trap.error;
        }
    }
    else if (type[0] == 'F') {
        int n = 0;
        int s = 0;
        int ch = read_stream(&c->inp);
        if (ch == '-') {
            s = 1;
            ch = read_stream(&c->inp);
        }
        while (ch >= '0' && ch <= '9') {
            n = n * 10 + (ch - '0');
            ch = read_stream(&c->inp);
        }
        if (ch == 0) {
            ch = read_stream(&c->inp);
        }
        else {
            trace(LOG_ALWAYS, "Received F with no zero termination.");
        }
        if (ch != MARKER_EOM) error = ERR_PROTOCOL;
        else c->congestion_level = s ? -n : n;
    }
    else if (p->default_handler != NULL) {
        args[0] = type;
        p->default_handler(c, args, 1, p->client_data);
    }
    else {
        trace(LOG_ALWAYS, "Invalid TCF message: %s ...", type);
        error = ERR_PROTOCOL;
    }
    if (error != 0) exception(error);
}

static void message_handler_old(Channel * c, char ** args, int nargs, void * client_data) {
    ProtocolMessageHandler handler = (ProtocolMessageHandler)client_data;
    handler(c, args, nargs);
}

void set_default_message_handler(Protocol * p, ProtocolMessageHandler handler) {
    set_default_message_handler2(p, (ProtocolMessageHandler2)message_handler_old, (void *)handler);
}

void set_default_message_handler2(Protocol * p, ProtocolMessageHandler2 handler, void * client_data) {
    p->default_handler = handler;
    p->client_data = client_data;
}

static void command_handler_old(char * token, Channel * c, void * client_data) {
    ProtocolCommandHandler handler = (ProtocolCommandHandler)client_data;
    handler(token, c);
}

void add_command_handler(Protocol * p, const char * service, const char * name, ProtocolCommandHandler handler) {
    add_command_handler2(p, service, name, (ProtocolCommandHandler2)command_handler_old, (void *)handler);
}

void add_command_handler2(Protocol * p, const char * service, const char * name, ProtocolCommandHandler2 handler, void * client_data) {
    unsigned h = message_hash(p, service, name);
    MessageHandlerInfo * mh = (MessageHandlerInfo *)loc_alloc(sizeof(MessageHandlerInfo));
    mh->p = p;
    mh->service = protocol_get_service(p, service);
    mh->name = name;
    mh->handler = handler;
    mh->client_data = client_data;
    mh->next = message_handlers[h];
    message_handlers[h] = mh;
}

static void event_handler_old(Channel * c, void * client_data) {
    ProtocolEventHandler handler = (ProtocolEventHandler)client_data;
    handler(c);
}

void add_event_handler(Channel * c, const char * service, const char * name, ProtocolEventHandler handler) {
    add_event_handler2(c, service, name, (ProtocolEventHandler2)event_handler_old, (void *)handler);
}

void add_event_handler2(Channel * c, const char * service, const char * name, ProtocolEventHandler2 handler, void * client_data) {
    unsigned h = event_hash(c, service, name);
    EventHandlerInfo * eh = (EventHandlerInfo *)loc_alloc(sizeof(EventHandlerInfo));
    eh->c = c;
    eh->service = protocol_get_service(c, service);
    eh->name = name;
    eh->handler = handler;
    eh->client_data = client_data;
    eh->next = event_handlers[h];
    event_handlers[h] = eh;
}

static void send_command_failed(void * args) {
    ReplyHandlerInfo * rh = (ReplyHandlerInfo *)args;
    rh->handler(rh->c, rh->client_data, ERR_CHANNEL_CLOSED);
    loc_free(rh);
}

ReplyHandlerInfo * protocol_send_command(Channel * c, const char * service, const char * name, ReplyHandlerCB handler, void * client_data) {
    Protocol * p = c->protocol;
    ReplyHandlerInfo * rh = (ReplyHandlerInfo *)loc_alloc(sizeof(ReplyHandlerInfo));

    rh->c = c;
    rh->handler = handler;
    rh->client_data = client_data;
    if (c->peer_service_list == NULL) {
        post_event(send_command_failed, rh);
    }
    else {
        unsigned h;
        unsigned long tokenid;
        do tokenid = p->tokenid++;
        while (find_reply_handler(c, tokenid, 0) != NULL);
        write_stringz(&c->out, "C");
        json_write_ulong(&c->out, tokenid);
        write_stream(&c->out, 0);
        write_stringz(&c->out, service);
        write_stringz(&c->out, name);
        rh->tokenid = tokenid;
        h = reply_hash(c, tokenid);
        rh->next = reply_handlers[h];
        reply_handlers[h] = rh;
    }
    return rh;
}

struct sendRedirectInfo {
    ReplyHandlerCB handler;
    void * client_data;
};

static void redirect_done(Channel * c, void * client_data, int error) {
    struct sendRedirectInfo * info = (struct sendRedirectInfo *)client_data;

    if (!error) {
        assert(c->state == ChannelStateRedirectSent);
        error = read_errno(&c->inp);
        if (read_stream(&c->inp) != MARKER_EOM) exception(ERR_JSON_SYNTAX);
        if (!error) {
            c->state = ChannelStateHelloSent;
        }
        else {
            c->state = ChannelStateConnected;
        }
    }
    else if (c->state == ChannelStateRedirectSent) {
        c->state = ChannelStateConnected;
    }
    else {
        assert(c->state == ChannelStateDisconnected);
    }
    info->handler(c, info->client_data, error);
}

ReplyHandlerInfo * send_redirect_command(Channel * c, const char * peerId, ReplyHandlerCB handler, void * client_data) {
    struct sendRedirectInfo * info = (struct sendRedirectInfo *)loc_alloc_zero(sizeof *info);
    ReplyHandlerInfo * rh;

    assert(c->state == ChannelStateConnected);
    c->state = ChannelStateRedirectSent;
    info->handler = handler;
    info->client_data = client_data;
    rh = protocol_send_command(c, LOCATOR, "redirect", redirect_done, info);
    json_write_string(&c->out, peerId);
    write_stream(&c->out, 0);
    write_stream(&c->out, MARKER_EOM);
    return rh;
}

static void connect_done(Channel * c) {
    assert(c->state == ChannelStateConnected);
    if (c->connected) {
        c->connected(c);
    }
    else {
        int i;
        trace(LOG_PROTOCOL, "channel server connected, remote services:");
        for (i = 0; i < c->peer_service_cnt; i++) {
            trace(LOG_PROTOCOL, "  %s", c->peer_service_list[i]);
        }
    }
}

void send_hello_message(Channel * c) {
    Protocol * p = c->protocol;
    ServiceInfo * s = services;
    int cnt = 0;

    assert(c->state == ChannelStateStarted || c->state == ChannelStateHelloReceived);
    write_stringz(&c->out, "E");
    write_stringz(&c->out, LOCATOR);
    write_stringz(&c->out, "Hello");
    write_stream(&c->out, '[');
#if ENABLE_ZeroCopy
    if (!c->disable_zero_copy) {
        json_write_string(&c->out, "ZeroCopy");
        cnt++;
    }
#endif
    while (s) {
        if (s->owner == p) {
            if (cnt != 0) write_stream(&c->out, ',');
            json_write_string(&c->out, s->name);
            cnt++;
        }
        s = s->next;
    }
    write_stream(&c->out, ']');
    write_stream(&c->out, 0);
    write_stream(&c->out, MARKER_EOM);
    if (c->state == ChannelStateStarted) {
        c->state = ChannelStateHelloSent;
    }
    else {
        c->state = ChannelStateConnected;
        connect_done(c);
    }
}

static void free_string_list(int cnt, char **list) {
    while (cnt > 0) loc_free(list[--cnt]);
    loc_free(list);
}

static void event_locator_hello(Channel * c) {
    int cnt = 0;
    char **list = NULL;

    c->out.supports_zero_copy = 0;
    if (read_stream(&c->inp) != '[') exception(ERR_PROTOCOL);
    if (peek_stream(&c->inp) == ']') {
        read_stream(&c->inp);
    }
    else {
        int max = 4;
        list = (char **)loc_alloc(max * sizeof *list);
        for (;;) {
            int ch;
            char * service = json_read_alloc_string(&c->inp);
            if (strcmp(service, "ZeroCopy") == 0) c->out.supports_zero_copy = 1;
            if (cnt == max) {
                max *= 2;
                list = (char **)loc_realloc(list, max * sizeof *list);
            }
            list[cnt++] = service;
            ch = read_stream(&c->inp);
            if (ch == ',') continue;
            if (ch == ']') break;
            free_string_list(cnt, list);
            exception(ERR_JSON_SYNTAX);
        }
    }
    if (read_stream(&c->inp) != 0 || read_stream(&c->inp) != MARKER_EOM) {
        free_string_list(cnt, list);
        exception(ERR_JSON_SYNTAX);
    }
    if (c->state != ChannelStateStarted && c->state != ChannelStateHelloSent) {
        free_string_list(cnt, list);
        /* TODO: should this be a protocol error? */
        return;
    }
    if (c->peer_service_list != NULL) {
        free_string_list(c->peer_service_cnt, c->peer_service_list);
    }
    c->peer_service_cnt = cnt;
    c->peer_service_list = list;
    if (c->state == ChannelStateStarted) {
        c->state = ChannelStateHelloReceived;
    }
    else {
        c->state = ChannelStateConnected;
        connect_done(c);
    }
}

int protocol_cancel_command(ReplyHandlerInfo * rh) {
    /* TODO: protocol_cancel_command() */
    return 0;
}

static void channel_closed(Channel * c) {
    unsigned i;

    assert(is_dispatch_thread());
    for (i = 0; i < EVENT_HASH_SIZE; i++) {
        EventHandlerInfo ** ehp = &event_handlers[i];
        EventHandlerInfo * eh;

        while ((eh = *ehp) != NULL) {
            if (eh->c == c) {
                *ehp = eh->next;
                loc_free(eh);
            }
            else {
                ehp = &eh->next;
            }
        }
    }
    free_services(c);

    for (i = 0; i < REPLY_HASH_SIZE; i++) {
        ReplyHandlerInfo ** rhp = &reply_handlers[i];
        ReplyHandlerInfo * rh;
        while ((rh = *rhp) != NULL) {
            if (rh->c == c) {
                Trap trap;
                *rhp = rh->next;
                if (set_trap(&trap)) {
                    rh->handler(c, rh->client_data, ERR_CHANNEL_CLOSED);
                    clear_trap(&trap);
                }
                else {
                    trace(LOG_ALWAYS, "Exception handling reply %ul: %d %s",
                          rh->tokenid, trap.error, errno_to_str(trap.error));
                }
                loc_free(rh);
            }
            else {
                rhp = &rh->next;
            }
        }
    }
    if (c->peer_service_list) {
        free_string_list(c->peer_service_cnt, c->peer_service_list);
        c->peer_service_cnt = 0;
        c->peer_service_list = NULL;
    }
}

static void ini_protocol(void) {
    assert(!ini_done);
    agent_id = loc_strdup(create_uuid());
    add_channel_close_listener(channel_closed);
    ini_done = 1;
}

Protocol * protocol_alloc(void) {
    Protocol * p = (Protocol *)loc_alloc_zero(sizeof *p);

    assert(is_dispatch_thread());
    if (!ini_done) ini_protocol();
    p->id = proto_cnt++;
    p->lock_cnt = 1;
    p->tokenid = 1;
    return p;
}

void protocol_reference(Protocol * p) {
    assert(is_dispatch_thread());
    assert(p->lock_cnt > 0);
    p->lock_cnt++;
}

void protocol_release(Protocol * p) {
    MessageHandlerInfo ** mhp;
    MessageHandlerInfo * mh;
    int i;

    assert(is_dispatch_thread());
    assert(p->lock_cnt > 0);
    if (--p->lock_cnt != 0) return;
    for (i = 0; i < MESSAGE_HASH_SIZE; i++) {
        mhp = &message_handlers[i];
        while ((mh = *mhp) != NULL) {
            if (mh->p == p) {
                *mhp = mh->next;
                loc_free(mh);
            }
            else {
                mhp = &mh->next;
            }
        }
    }
    free_services(p);
}

const char * get_agent_id(void) {
    if (!ini_done) ini_protocol();
    return agent_id;
}

const char * get_service_manager_id(Protocol * p) {
    static char buf[256];
    snprintf(buf, sizeof(buf), "%s-%d", get_agent_id(), p->id);
    return buf;
}