diff --git a/core/od.c b/core/od.c index 7427c903..7a98b2e6 100644 --- a/core/od.c +++ b/core/od.c @@ -11,6 +11,7 @@ #include #include +#include #include "od_macro.h" #include "od_list.h" @@ -88,7 +89,6 @@ int od_main(od_t *od, int argc, char **argv) rc = od_schemevalidate(&od->scheme, &od->log); if (rc == -1) return 1; - od_log(&od->log, "ready."); /* run connection pooler */ odpooler_t pooler; rc = od_pooler_init(&pooler, od); diff --git a/core/od_client.h b/core/od_client.h index 141b0e6c..6e147a0f 100644 --- a/core/od_client.h +++ b/core/od_client.h @@ -10,10 +10,11 @@ typedef struct odclient_t odclient_t; struct odclient_t { - ftio_t io; - odserver_t *server; - void *pooler; - odlist_t link; + ftio_t io; + sostream_t stream; + odserver_t *server; + void *pooler; + odlist_t link; }; static inline void @@ -22,6 +23,7 @@ od_clientinit(odclient_t *c) c->io = NULL; c->server = NULL; c->pooler = NULL; + so_stream_init(&c->stream); od_listinit(&c->link); } @@ -38,6 +40,7 @@ od_clientalloc(void) static inline void od_clientfree(odclient_t *c) { + so_stream_free(&c->stream); free(c); } diff --git a/core/od_client_pool.c b/core/od_client_pool.c index 6a46e9db..b7f29e4c 100644 --- a/core/od_client_pool.c +++ b/core/od_client_pool.c @@ -12,6 +12,7 @@ #include #include +#include #include "od_macro.h" #include "od_list.h" diff --git a/core/od_pooler.c b/core/od_pooler.c index 82867abc..c7ddc8b6 100644 --- a/core/od_pooler.c +++ b/core/od_pooler.c @@ -11,6 +11,7 @@ #include #include +#include #include "od_macro.h" #include "od_list.h" @@ -42,6 +43,8 @@ od_pooler(void *arg) env->scheme.port); return; } + od_log(&env->log, "pooler started at %s:%d", + env->scheme.host, env->scheme.port); /* accept loop */ while (ft_is_online(pooler->env)) diff --git a/core/od_router.c b/core/od_router.c index 10b023bd..e4253373 100644 --- a/core/od_router.c +++ b/core/od_router.c @@ -11,6 +11,7 @@ #include #include +#include #include "od_macro.h" #include "od_list.h" @@ -26,20 +27,141 @@ #include "od_pooler.h" #include "od_router.h" +static void +od_router_close(odclient_t *client) +{ + odpooler_t *pooler = client->pooler; + if (client->io) { + ft_close(client->io); + client->io = NULL; + } + od_clientpool_unlink(&pooler->client_pool, client); +} + +static inline int +od_router_read(odclient_t *client) +{ + sostream_t *stream = &client->stream; + so_stream_reset(stream); + for (;;) { + uint32_t pos_size = so_stream_used(stream); + uint8_t *pos_data = stream->s; + uint32_t len; + int to_read; + to_read = so_read(&len, &pos_data, &pos_size); + if (to_read == 0) + break; + if (to_read == -1) + return -1; + int rc = so_stream_ensure(stream, to_read); + if (rc == -1) + return -1; + rc = ft_read(client->io, to_read, 0); + if (rc < 0) + return -1; + char *data_pointer = ft_read_buf(client->io); + memcpy(stream->p, data_pointer, to_read); + so_stream_advance(stream, to_read); + } + return 0; +} + +static inline int +od_router_startup_read(odclient_t *client) +{ + sostream_t *stream = &client->stream; + so_stream_reset(stream); + for (;;) { + uint32_t pos_size = so_stream_used(stream); + uint8_t *pos_data = stream->s; + uint32_t len; + int to_read; + to_read = so_read_startup(&len, &pos_data, &pos_size); + if (to_read == 0) + break; + if (to_read == -1) + return -1; + int rc = so_stream_ensure(stream, to_read); + if (rc == -1) + return -1; + rc = ft_read(client->io, to_read, 0); + if (rc < 0) + return -1; + char *data_pointer = ft_read_buf(client->io); + memcpy(stream->p, data_pointer, to_read); + so_stream_advance(stream, to_read); + } + return 0; +} + +static int +od_router_startup(odclient_t *client, sobestartup_t *startup) +{ + int rc; + rc = od_router_startup_read(client); + if (rc == -1) + return -1; + sostream_t *stream = &client->stream; + rc = so_beread_startup(startup, stream->s, so_stream_used(stream)); + if (rc == -1) + return -1; + return 0; +} + +static int +od_router_auth(odclient_t *client) +{ + sostream_t *stream = &client->stream; + so_stream_reset(stream); + int rc; + rc = so_bewrite_authentication(stream, 0); + rc = so_bewrite_backend_key_data(stream, 0, 0); + rc = so_bewrite_parameter_status(stream, "", 1, "", 1); + rc = so_bewrite_ready(stream, 'I'); + rc = ft_write(client->io, (char*)stream->s, so_stream_used(stream), 0); + return rc; +} + void od_router(void *arg) { odclient_t *client = arg; odpooler_t *pooler = client->pooler; - (void)pooler; - /* client buffer */ + od_log(&pooler->od->log, "C: new connection"); + + /* client startup */ + sobestartup_t startup; + memset(&startup, 0, sizeof(startup)); + int rc = od_router_startup(client, &startup); + if (rc == -1) { + od_router_close(client); + return; + } + /* client cancel request */ + if (startup.is_cancel) { + od_log(&pooler->od->log, "C: cancel request"); + od_router_close(client); + return; + } /* client auth */ + rc = od_router_auth(client); + if (rc == -1) { + od_router_close(client); + return; + } /* server = serverpool_pop() */ /* server = server_connect() */ while (1) { - /* packet = read(client) */ + rc = od_router_read(client); + if (rc == -1) { + od_router_close(client); + return; + } + char type = *client->stream.s; + od_log(&pooler->od->log, "C: %c", type); + /* write(server, packet) */ while (1) { /* packet = read(server) */ diff --git a/lib/soprano b/lib/soprano index 34951e25..fbe220af 160000 --- a/lib/soprano +++ b/lib/soprano @@ -1 +1 @@ -Subproject commit 34951e25e25e0717b0fd676a0249811a51bc99e4 +Subproject commit fbe220af315f878f9b01ad384d7b3a46be6a2559 diff --git a/odissey.conf b/odissey.conf index ab94c03f..7375b74a 100644 --- a/odissey.conf +++ b/odissey.conf @@ -11,8 +11,8 @@ odissey { pooling "session" # transaction, statement listen { - host "localhost" - port 1234 + host "127.0.0.1" + port 6432 # tls_sslmode disable # tls_ca_file # tls_key_file @@ -24,7 +24,7 @@ odissey { } server "default" { - host "localhost" + host "127.0.0.1" port 4321 # tls_sslmode disable # tls_ca_file