#include "ioloop.h"
#include "istream.h"
#include "ostream.h"
+#include "iostream.h"
+#include "iostream-proxy.h"
#include "llist.h"
#include "array.h"
#include "str.h"
#include "strescape.h"
-#include "str-sanitize.h"
#include "time-util.h"
#include "master-service.h"
#include "ipc-server.h"
#define MAX_PROXY_INPUT_SIZE 4096
-#define OUTBUF_THRESHOLD 1024
+#define PROXY_MAX_OUTBUF_SIZE 1024
#define LOGIN_PROXY_DIE_IDLE_SECS 2
#define LOGIN_PROXY_IPC_PATH "ipc-proxy"
#define LOGIN_PROXY_IPC_NAME "proxy"
#define PROXY_CONNECT_RETRY_MSECS 1000
#define PROXY_DISCONNECT_INTERVAL_MSECS 100
+#define LOGIN_PROXY_SIDE_CLIENT IOSTREAM_PROXY_SIDE_LEFT
+#define LOGIN_PROXY_SIDE_SERVER IOSTREAM_PROXY_SIDE_RIGHT
+
+
struct login_proxy {
struct login_proxy *prev, *next;
struct client *client;
int server_fd;
- struct io *client_io, *server_io;
+ struct io *client_wait_io, *server_io;
struct istream *client_input, *server_input;
struct ostream *client_output, *server_output;
+ struct iostream_proxy *iostream_proxy;
struct ssl_proxy *ssl_server_proxy;
- time_t last_io;
struct timeval created;
struct timeout *to, *to_notify;
login_proxy_free_delayed(struct login_proxy **_proxy, const char *reason)
ATTR_NULL(2);
+static time_t proxy_last_io(struct login_proxy *proxy)
+{
+ struct timeval tv1, tv2, tv3, tv4;
+
+ i_stream_get_last_read_time(proxy->client_input, &tv1);
+ i_stream_get_last_read_time(proxy->server_input, &tv2);
+ o_stream_get_last_write_time(proxy->client_output, &tv3);
+ o_stream_get_last_write_time(proxy->server_output, &tv4);
+ return I_MAX(tv1.tv_sec, I_MAX(tv2.tv_sec, I_MAX(tv3.tv_sec, tv4.tv_sec)));
+}
+
static void login_proxy_free_errstr(struct login_proxy **_proxy,
const char *errstr, bool server)
{
str_printfa(reason, ": %s", errstr);
str_printfa(reason, "(%ds idle, in=%"PRIuUOFF_T", out=%"PRIuUOFF_T,
- (int)(ioloop_time - proxy->last_io),
+ (int)(ioloop_time - proxy_last_io(proxy)),
proxy->server_output->offset, proxy->client_output->offset);
if (o_stream_get_buffer_used_size(proxy->client_output) > 0) {
str_printfa(reason, "+%"PRIuSIZE_T,
o_stream_get_buffer_used_size(proxy->client_output));
}
- if (proxy->server_io == NULL)
+ if (iostream_proxy_is_waiting_output(proxy->iostream_proxy,
+ LOGIN_PROXY_SIDE_SERVER))
str_append(reason, ", client output blocked");
- if (proxy->client_io == NULL)
+ if (iostream_proxy_is_waiting_output(proxy->iostream_proxy,
+ LOGIN_PROXY_SIDE_CLIENT))
str_append(reason, ", server output blocked");
+
str_append_c(reason, ')');
if (server)
login_proxy_free_delayed(_proxy, str_c(reason));
login_proxy_free_reason(_proxy, str_c(reason));
}
-static void login_proxy_free_errno(struct login_proxy **_proxy,
- int err, bool server)
-{
- const char *errstr;
-
- errstr = err == 0 || err == EPIPE ? "" : strerror(err);
- login_proxy_free_errstr(_proxy, errstr, server);
-}
-
-static void login_proxy_free_ostream(struct login_proxy **_proxy,
- struct ostream *output, bool server)
-{
- const char *errstr;
-
- errstr = output->stream_errno == 0 ||
- output->stream_errno == EPIPE ? "" :
- o_stream_get_error(output);
- login_proxy_free_errstr(_proxy, errstr, server);
-}
-
-static void server_input(struct login_proxy *proxy)
-{
- unsigned char buf[OUTBUF_THRESHOLD];
- ssize_t ret, ret2;
-
- proxy->last_io = ioloop_time;
- if (o_stream_get_buffer_used_size(proxy->client_output) >
- OUTBUF_THRESHOLD) {
- /* client's output buffer is already quite full.
- don't send more until we're below threshold. */
- io_remove(&proxy->server_io);
- return;
- }
-
- ret = net_receive(proxy->server_fd, buf, sizeof(buf));
- if (ret < 0) {
- login_proxy_free_errno(&proxy, errno, TRUE);
- return;
- }
- o_stream_cork(proxy->client_output);
- ret2 = o_stream_send(proxy->client_output, buf, ret);
- o_stream_uncork(proxy->client_output);
- if (ret2 != ret)
- login_proxy_free_ostream(&proxy, proxy->client_output, FALSE);
-}
-
-static void proxy_client_input(struct login_proxy *proxy)
-{
- const unsigned char *data;
- size_t size;
- ssize_t ret;
-
- proxy->last_io = ioloop_time;
- if (o_stream_get_buffer_used_size(proxy->server_output) >
- OUTBUF_THRESHOLD) {
- /* proxy's output buffer is already quite full.
- don't send more until we're below threshold. */
- io_remove(&proxy->client_io);
- return;
- }
-
- if (i_stream_read_more(proxy->client_input, &data, &size) < 0) {
- const char *errstr = i_stream_get_error(proxy->client_input);
- login_proxy_free_errstr(&proxy, errstr, FALSE);
- return;
- }
- o_stream_cork(proxy->server_output);
- ret = o_stream_send(proxy->server_output, data, size);
- o_stream_uncork(proxy->server_output);
- if (ret != (ssize_t)size)
- login_proxy_free_ostream(&proxy, proxy->server_output, TRUE);
- else
- i_stream_skip(proxy->client_input, ret);
-}
-
static void proxy_client_disconnected_input(struct login_proxy *proxy)
{
/* we're already disconnected from server. either wait for
}
}
-static int server_output(struct login_proxy *proxy)
-{
- proxy->last_io = ioloop_time;
- if (o_stream_flush(proxy->server_output) < 0) {
- login_proxy_free_ostream(&proxy, proxy->server_output, TRUE);
- return 1;
- }
-
- if (proxy->client_io == NULL &&
- o_stream_get_buffer_used_size(proxy->server_output) <
- OUTBUF_THRESHOLD) {
- /* there's again space in proxy's output buffer, so we can
- read more from client. */
- proxy->client_io = io_add_istream(proxy->client_input,
- proxy_client_input, proxy);
- }
- return 1;
-}
-
-static int proxy_client_output(struct login_proxy *proxy)
-{
- proxy->last_io = ioloop_time;
- if (o_stream_flush(proxy->client_output) < 0) {
- login_proxy_free_ostream(&proxy, proxy->client_output, FALSE);
- return 1;
- }
-
- if (proxy->server_io == NULL &&
- o_stream_get_buffer_used_size(proxy->client_output) <
- OUTBUF_THRESHOLD) {
- /* there's again space in client's output buffer, so we can
- read more from proxy. */
- proxy->server_io =
- io_add(proxy->server_fd, IO_READ, server_input, proxy);
- }
- return 1;
-}
-
static void proxy_prelogin_input(struct login_proxy *proxy)
{
proxy->callback(proxy->client);
proxy->state_rec->num_proxying_connections--;
}
+ if (proxy->iostream_proxy != NULL)
+ iostream_proxy_unref(&proxy->iostream_proxy);
io_remove(&proxy->server_io);
i_stream_destroy(&proxy->server_input);
o_stream_destroy(&proxy->server_output);
timeout_remove(&proxy->to);
}
- io_remove(&proxy->client_io);
+ io_remove(&proxy->client_wait_io);
i_stream_destroy(&proxy->client_input);
o_stream_destroy(&proxy->client_output);
if (proxy->ssl_server_proxy != NULL) {
ipstr != NULL ? ipstr : "",
reason == NULL ? "" : t_strdup_printf(" (%s)", reason),
delay_ms == 0 ? "" : t_strdup_printf(" - disconnecting client in %ums", delay_ms)));
-
- if (proxy->client_io != NULL)
- io_remove(&proxy->client_io);
} else {
- i_assert(proxy->client_io == NULL);
i_assert(proxy->client_input == NULL);
i_assert(proxy->client_output == NULL);
if (delay_ms == 0)
login_proxy_free_final(proxy);
else {
- proxy->client_io = io_add_istream(proxy->client_input,
+ i_assert(proxy->client_wait_io == NULL);
+ proxy->client_wait_io = io_add_istream(proxy->client_input,
proxy_client_disconnected_input, proxy);
}
return proxy->ssl_flags;
}
+static void
+login_proxy_finished(enum iostream_proxy_side side,
+ enum iostream_proxy_status status,
+ struct login_proxy *proxy)
+{
+ const char *errstr;
+ bool server_side;
+
+ server_side = side == LOGIN_PROXY_SIDE_SERVER;
+ switch (status) {
+ case IOSTREAM_PROXY_STATUS_INPUT_EOF:
+ /* success */
+ errstr = "";
+ break;
+ case IOSTREAM_PROXY_STATUS_INPUT_ERROR:
+ errstr = side == LOGIN_PROXY_SIDE_CLIENT ?
+ i_stream_get_error(proxy->client_input) :
+ i_stream_get_error(proxy->server_input);
+ break;
+ case IOSTREAM_PROXY_STATUS_OTHER_SIDE_OUTPUT_ERROR:
+ server_side = !server_side;
+ errstr = side == LOGIN_PROXY_SIDE_CLIENT ?
+ o_stream_get_error(proxy->server_output) :
+ o_stream_get_error(proxy->client_output);
+ break;
+ default:
+ i_unreached();
+ }
+ login_proxy_free_errstr(&proxy, errstr, server_side);
+}
+
static void login_proxy_notify(struct login_proxy *proxy)
{
login_proxy_state_notify(proxy_state, proxy->client->proxy_user);
void login_proxy_detach(struct login_proxy *proxy)
{
struct client *client = proxy->client;
- const unsigned char *data;
- size_t size;
pool_unref(&proxy->client->preproxy_pool);
i_assert(proxy->server_output != NULL);
timeout_remove(&proxy->to);
+ io_remove(&proxy->server_io);
proxy->detached = TRUE;
proxy->client_input = client->input;
proxy->client_output = client->output;
+ i_stream_set_persistent_buffers(proxy->server_input, FALSE);
i_stream_set_persistent_buffers(client->input, FALSE);
- o_stream_set_max_buffer_size(client->output, (size_t)-1);
- o_stream_set_flush_callback(client->output, proxy_client_output, proxy);
+ o_stream_set_max_buffer_size(client->output, PROXY_MAX_OUTBUF_SIZE);
client->input = NULL;
client->output = NULL;
- /* send all pending client input to proxy */
- data = i_stream_get_data(proxy->client_input, &size);
- if (size != 0)
- o_stream_nsend(proxy->server_output, data, size);
-
/* from now on, just do dummy proxying */
- io_remove(&proxy->server_io);
- proxy->server_io =
- io_add(proxy->server_fd, IO_READ, server_input, proxy);
- proxy->client_io =
- io_add_istream(proxy->client_input, proxy_client_input, proxy);
- o_stream_set_flush_callback(proxy->server_output, server_output, proxy);
- i_stream_destroy(&proxy->server_input);
+ proxy->iostream_proxy =
+ iostream_proxy_create(proxy->client_input, proxy->client_output,
+ proxy->server_input, proxy->server_output);
+ iostream_proxy_set_completion_callback(proxy->iostream_proxy,
+ login_proxy_finished, proxy);
+ iostream_proxy_start(proxy->iostream_proxy);
if (proxy->notify_refresh_secs != 0) {
proxy->to_notify =
for (proxy = login_proxies; proxy != NULL; proxy = next) {
next = proxy->next;
+ time_t last_io = proxy_last_io(proxy);
- if (proxy->last_io <= stop_timestamp)
+ if (last_io <= stop_timestamp)
proxy_kill_idle(proxy);
else {
i_assert(proxy->to == NULL);
- stop_msecs = (proxy->last_io - stop_timestamp) * 1000;
+ stop_msecs = (last_io - stop_timestamp) * 1000;
proxy->to = timeout_add(stop_msecs,
proxy_kill_idle, proxy);
}
projects / thirdparty / dovecot / core.git / commitdiff
