Lines Matching refs:conn
46 struct worker_connection *conn;
48 conn = i_new(struct worker_connection, 1);
49 conn->refcount = 1;
50 conn->socket_path = i_strdup(socket_path);
51 conn->callback = callback;
52 conn->fd = -1;
53 i_array_init(&conn->request_contexts, 32);
54 conn->request_queue = aqueue_init(&conn->request_contexts.arr);
55 return conn;
58 static void worker_connection_unref(struct worker_connection *conn)
60 i_assert(conn->refcount > 0);
61 if (--conn->refcount > 0)
64 aqueue_deinit(&conn->request_queue);
65 array_free(&conn->request_contexts);
66 i_free(conn->socket_path);
67 i_free(conn);
70 static void worker_connection_disconnect(struct worker_connection *conn)
72 unsigned int i, count = aqueue_count(conn->request_queue);
74 if (conn->fd != -1) {
75 io_remove(&conn->io);
76 i_stream_destroy(&conn->input);
77 o_stream_destroy(&conn->output);
79 if (close(conn->fd) < 0)
80 i_error("close(%s) failed: %m", conn->socket_path);
81 conn->fd = -1;
88 count, conn->request_username);
91 /* conn->callback() can try to destroy us */
92 conn->refcount++;
95 array_idx(&conn->request_contexts,
96 aqueue_idx(conn->request_queue, 0));
99 aqueue_delete_tail(conn->request_queue);
100 conn->callback(-1, context);
102 i_free_and_null(conn->request_username);
103 worker_connection_unref(conn);
108 struct worker_connection *conn = *_conn;
112 worker_connection_disconnect(conn);
113 worker_connection_unref(conn);
117 worker_connection_input_line(struct worker_connection *conn, const char *line)
122 if (aqueue_count(conn->request_queue) == 0) {
133 contextp = array_idx(&conn->request_contexts,
134 aqueue_idx(conn->request_queue, 0));
138 aqueue_delete_tail(conn->request_queue);
139 if (aqueue_count(conn->request_queue) == 0)
140 i_free_and_null(conn->request_username);
143 conn->callback(percentage, context);
147 static void worker_connection_input(struct worker_connection *conn)
151 if (i_stream_read(conn->input) < 0) {
152 worker_connection_disconnect(conn);
156 if (!conn->version_received) {
157 if ((line = i_stream_next_line(conn->input)) == NULL)
164 worker_connection_disconnect(conn);
167 conn->version_received = TRUE;
169 if (conn->process_limit == 0) {
170 if ((line = i_stream_next_line(conn->input)) == NULL)
172 if (str_to_uint(line, &conn->process_limit) < 0 ||
173 conn->process_limit == 0) {
176 worker_connection_disconnect(conn);
181 while ((line = i_stream_next_line(conn->input)) != NULL) {
182 if (worker_connection_input_line(conn, line) < 0) {
183 worker_connection_disconnect(conn);
189 int worker_connection_connect(struct worker_connection *conn)
191 i_assert(conn->fd == -1);
193 conn->fd = net_connect_unix(conn->socket_path);
194 if (conn->fd == -1) {
195 i_error("connect(%s) failed: %m", conn->socket_path);
198 conn->io = io_add(conn->fd, IO_READ, worker_connection_input, conn);
199 conn->input = i_stream_create_fd(conn->fd, (size_t)-1);
200 conn->output = o_stream_create_fd(conn->fd, (size_t)-1);
201 o_stream_set_no_error_handling(conn->output, TRUE);
202 o_stream_nsend_str(conn->output, INDEXER_MASTER_HANDSHAKE);
206 bool worker_connection_is_connected(struct worker_connection *conn)
208 return conn->fd != -1;
211 bool worker_connection_get_process_limit(struct worker_connection *conn,
214 if (conn->process_limit == 0)
217 *limit_r = conn->process_limit;
221 void worker_connection_request(struct worker_connection *conn,
225 i_assert(worker_connection_is_connected(conn));
229 if (conn->request_username == NULL)
230 conn->request_username = i_strdup(request->username);
232 i_assert(strcmp(conn->request_username,
236 aqueue_append(conn->request_queue, &context);
253 o_stream_nsend(conn->output, str_data(str), str_len(str));
257 bool worker_connection_is_busy(struct worker_connection *conn)
259 return aqueue_count(conn->request_queue) > 0;
262 const char *worker_connection_get_username(struct worker_connection *conn)
264 return conn->request_username;