- while (msg_list.head) {
- struct msg_list_item *ml = msg_list.head;
- int n = write(msg_fd_out, ml->buf + written, ml->len - written);
- if (n < 0) {
- if (errno == EINTR)
- continue;
- if (errno != EWOULDBLOCK && errno != EAGAIN)
- return -1;
- if (!flush_it_all)
- return 0;
- FD_ZERO(&fds);
- FD_SET(msg_fd_out, &fds);
- tv.tv_sec = select_timeout;
- tv.tv_usec = 0;
- if (!select(msg_fd_out+1, NULL, &fds, NULL, &tv))
- check_timeout();
- } else if ((written += n) == ml->len) {
- free(ml->buf);
- msg_list.head = ml->next;
- if (!msg_list.head)
- msg_list.tail = NULL;
- free(ml);
- written = 0;
- }
+/* Write an message to a multiplexed stream. If this fails, rsync exits. */
+static void mplex_write(int fd, enum msgcode code, const char *buf, size_t len, int convert)
+{
+ char buffer[BIGPATHBUFLEN]; /* Oversized for use by iconv code. */
+ size_t n = len;
+
+ SIVAL(buffer, 0, ((MPLEX_BASE + (int)code)<<24) + len);
+
+#ifdef ICONV_OPTION
+ if (convert && ic_send == (iconv_t)-1)
+#endif
+ convert = 0;
+
+ if (convert || n > 1024 - 4) /* BIGPATHBUFLEN can handle 1024 bytes */
+ n = 0;
+ else
+ memcpy(buffer + 4, buf, n);
+
+ writefd_unbuffered(fd, buffer, n+4);
+
+ len -= n;
+ buf += n;
+
+#ifdef ICONV_OPTION
+ if (convert) {
+ xbuf outbuf, inbuf;
+
+ INIT_CONST_XBUF(outbuf, buffer);
+ INIT_XBUF(inbuf, (char*)buf, len, -1);
+
+ defer_forwarding_messages++;
+ do {
+ iconvbufs(ic_send, &inbuf, &outbuf,
+ ICB_INCLUDE_BAD | ICB_INCLUDE_INCOMPLETE);
+ writefd_unbuffered(fd, outbuf.buf, outbuf.len);
+ } while (inbuf.len);
+ if (!--defer_forwarding_messages)
+ msg_flush();
+ } else
+#endif
+ if (len) {
+ defer_forwarding_messages++;
+ writefd_unbuffered(fd, buf, len);
+ if (!--defer_forwarding_messages)
+ msg_flush();
+ }
+}
+
+int send_msg(enum msgcode code, const char *buf, int len, int convert)
+{
+ if (msg_fd_out < 0) {
+ if (!defer_forwarding_messages)
+ return io_multiplex_write(code, buf, len, convert);
+ if (!io_multiplexing_out)
+ return 0;
+ msg_list_add(&msg_queue, code, buf, len, convert);
+ return 1;