1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
/* Copyright (c) 2013-2018 Dovecot authors, see the included COPYING file */
#include "lib.h"
#include "ioloop.h"
#include "net.h"
#include "istream.h"
#include "ostream.h"
#include "llist.h"
#include "strescape.h"
#include "master-service.h"
#include "replication-common.h"
#include "replicator-connection.h"
#include "notify-connection.h"
#define MAX_INBUF_SIZE 8192
#define CONNECTION_IS_FIFO(conn) \
((conn)->output == NULL)
struct notify_connection {
struct notify_connection *prev, *next;
int refcount;
int fd;
struct io *io;
struct istream *input;
struct ostream *output;
};
static struct notify_connection *conns = NULL;
static void notify_connection_unref(struct notify_connection *conn);
static void notify_connection_destroy(struct notify_connection *conn);
static bool notify_input_error(struct notify_connection *conn)
{
if (CONNECTION_IS_FIFO(conn))
return TRUE;
notify_connection_destroy(conn);
return FALSE;
}
void notify_connection_sync_callback(bool success, void *context)
{
struct notify_connection *conn = context;
o_stream_nsend_str(conn->output, success ? "+\n" : "-\n");
notify_connection_unref(conn);
}
static int
notify_input_line(struct notify_connection *conn, const char *line)
{
const char *const *args;
enum replication_priority priority;
/* <username> \t <priority> */
args = t_strsplit_tabescaped(line);
if (str_array_length(args) < 2) {
i_error("Client sent invalid input");
return -1;
}
if (replication_priority_parse(args[1], &priority) < 0) {
i_error("Client sent invalid priority: %s", args[1]);
return -1;
}
if (priority != REPLICATION_PRIORITY_SYNC)
replicator_connection_notify(replicator, args[0], priority);
else {
conn->refcount++;
replicator_connection_notify_sync(replicator, args[0], conn);
}
return 0;
}
static void notify_input(struct notify_connection *conn)
{
const char *line;
int ret;
switch (i_stream_read(conn->input)) {
case -2:
/* buffer full */
i_error("Client sent too long line");
(void)notify_input_error(conn);
return;
case -1:
/* disconnected */
notify_connection_destroy(conn);
return;
}
while ((line = i_stream_next_line(conn->input)) != NULL) {
T_BEGIN {
ret = notify_input_line(conn, line);
} T_END;
if (ret < 0) {
if (!notify_input_error(conn))
return;
}
}
}
void notify_connection_create(int fd, bool fifo)
{
struct notify_connection *conn;
conn = i_new(struct notify_connection, 1);
conn->refcount = 1;
conn->fd = fd;
conn->io = io_add(fd, IO_READ, notify_input, conn);
conn->input = i_stream_create_fd(fd, MAX_INBUF_SIZE);
if (!fifo) {
conn->output = o_stream_create_fd(fd, SIZE_MAX);
o_stream_set_no_error_handling(conn->output, TRUE);
}
DLLIST_PREPEND(&conns, conn);
}
static void notify_connection_unref(struct notify_connection *conn)
{
i_assert(conn->refcount > 0);
if (--conn->refcount > 0)
return;
i_stream_destroy(&conn->input);
o_stream_destroy(&conn->output);
i_free(conn);
}
static void notify_connection_destroy(struct notify_connection *conn)
{
i_assert(conn->fd != -1);
if (!CONNECTION_IS_FIFO(conn))
master_service_client_connection_destroyed(master_service);
DLLIST_REMOVE(&conns, conn);
io_remove(&conn->io);
i_stream_close(conn->input);
o_stream_close(conn->output);
net_disconnect(conn->fd);
conn->fd = -1;
notify_connection_unref(conn);
}
void notify_connections_destroy_all(void)
{
while (conns != NULL)
notify_connection_destroy(conns);
}
|
