#include "jabberd.h"
typedef enum { conn_DIE, conn_CLOSED, conn_OPEN, conn_AUTHD } conn_state;
typedef struct
{
mio io;
conn_state state;
char* hostip;
u_short hostport;
char* secret;
int timeout;
int tries_left;
pool mempool;
instance inst;
pth_msgport_t write_queue;
dpacket dplast;
} *conn_info, _conn_info;
typedef struct
{
pth_message_t head;
dpacket packet;
} *conn_write_buf, _conn_write_buf;
void base_connect_process_xml(mio m, int state, void* arg, xmlnode x);
result base_connect_deliver(instance i, dpacket p, void* arg)
{
conn_info ci = (conn_info)arg;
if (ci->state != conn_AUTHD)
{
conn_write_buf entry = pmalloco(p->p, sizeof(_conn_write_buf));
entry->packet = p;
pth_msgport_put(ci->write_queue, (pth_message_t*)entry);
}
else
{
if(ci->dplast == p)
deliver_fail(p,"Circular Reference Detected");
else
mio_write(ci->io, p->x, NULL, 0);
}
return r_DONE;
}
void base_connect_connect(void *arg)
{
conn_info ci = (conn_info)arg;
pth_sleep(2);
mio_connect(ci->hostip, ci->hostport, base_connect_process_xml, (void*)ci, ci->timeout, NULL, mio_handlers_new(NULL, NULL, MIO_XML_PARSER));
}
void base_connect_process_xml(mio m, int state, void* arg, xmlnode x)
{
conn_info ci = (conn_info)arg;
xmlnode cur;
char hashbuf[41];
log_debug(ZONE, "process XML: m:%X state:%d, arg:%X, x:%X", m, state, arg, x);
switch (state)
{
case MIO_NEW:
ci->state = conn_OPEN;
ci->io = m;
log_debug(ZONE, "base_connecting: %X, %X, %s", ci, ci->inst, ci->inst->id);
cur = xstream_header("jabber:component:accept", ci->inst->id, NULL);
mio_write(m, NULL, xstream_header_char(cur), -1);
xmlnode_free(cur);
return;
case MIO_XML_ROOT:
shahash_r(spools(x->p, xmlnode_get_attrib(x, "id"), ci->secret, x->p), hashbuf);
cur = xmlnode_new_tag("handshake");
xmlnode_insert_cdata(cur, hashbuf, -1);
mio_write(m, NULL, xmlnode2str(cur), -1);
xmlnode_free(cur);
xmlnode_free(x);
return;
case MIO_XML_NODE:
if (ci->state == conn_AUTHD)
{
ci->dplast = dpacket_new(x);
deliver(ci->dplast, ci->inst);
ci->dplast = NULL;
return;
}
if (j_strcmp(xmlnode_get_name(x), "handshake") == 0)
{
conn_write_buf b;
while ((b = (conn_write_buf) pth_msgport_get(ci->write_queue)) != NULL)
mio_write(ci->io, b->packet->x, NULL, 0);
ci->state = conn_AUTHD;
}
xmlnode_free(x);
return;
case MIO_CLOSED:
if(ci->state == conn_DIE)
return;
ci->state = conn_CLOSED;
if(ci->tries_left != -1)
ci->tries_left--;
if(ci->tries_left == 0)
{
fprintf(stderr, "Base Connect Failed: service %s was unable to connect to %s:%d, unrecoverable error, exiting", ci->inst->id, ci->hostip, ci->hostport);
exit(1);
}
log_debug(ZONE, "Base Connect Failed to connect to %s:%d Retry [%d] in 2 seconds...", ci->hostip, ci->hostport, ci->tries_left);
mtq_send(NULL,ci->mempool,base_connect_connect,(void *)ci);
return;
}
}
void base_connect_kill(void *arg)
{
conn_info ci = (conn_info)arg;
ci->state = conn_DIE;
}
result base_connect_config(instance id, xmlnode x, void *arg)
{
char* secret = NULL;
int timeout;
int tries;
int port;
conn_info ci = NULL;
port = j_atoi(xmlnode_get_tag_data(x, "port"),0);
secret = xmlnode_get_tag_data(x, "secret");
timeout = j_atoi(xmlnode_get_tag_data(x, "timeout"), 5);
tries = j_atoi(xmlnode_get_tag_data(x, "tries"), -1);
if(id == NULL)
{
log_debug(ZONE,"base_accept_config validating configuration\n");
if(port == 0 || (secret == NULL))
{
xmlnode_put_attrib(x, "error", "<connect> requires the following subtags: <port>, and <secret>");
return r_ERR;
}
return r_PASS;
}
log_debug(ZONE, "Activating configuration: %s\n", xmlnode2str(x));
ci = pmalloco(id->p, sizeof(_conn_info));
ci->mempool = id->p;
ci->state = conn_CLOSED;
ci->inst = id;
ci->hostip = pstrdup(ci->mempool, xmlnode_get_tag_data(x,"ip"));
if(ci->hostip == NULL) ci->hostip = pstrdup(ci->mempool, "127.0.0.1");
ci->hostport = port;
ci->secret = pstrdup(ci->mempool, secret);
ci->write_queue = pth_msgport_create(ci->hostip);
ci->timeout = timeout;
ci->tries_left = tries;
register_phandler(id, o_DELIVER, base_connect_deliver, (void*)ci);
mtq_send(NULL,ci->mempool,base_connect_connect,(void *)ci);
register_shutdown(base_connect_kill, (void *)ci);
return r_DONE;
}
void base_connect(void)
{
log_debug(ZONE,"base_connect loading...\n");
register_config("connect",base_connect_config,NULL);
}