+ switch (tag) {
+ case MSG_DONE:
+ if (len != 0 || !am_generator) {
+ rprintf(FERROR, "invalid message %d:%d\n", tag, len);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ flist_ndx_push(&redo_list, -1);
+ break;
+ case MSG_REDO:
+ if (len != 4 || !am_generator) {
+ rprintf(FERROR, "invalid message %d:%d\n", tag, len);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ read_loop(fd, buf, 4);
+ if (remove_source_files)
+ decrement_active_files(IVAL(buf,0));
+ flist_ndx_push(&redo_list, IVAL(buf,0));
+ break;
+ case MSG_DELETED:
+ if (len >= (int)sizeof buf || !am_generator) {
+ rprintf(FERROR, "invalid message %d:%d\n", tag, len);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ read_loop(fd, buf, len);
+ send_msg(MSG_DELETED, buf, len);
+ break;
+ case MSG_SUCCESS:
+ if (len != 4 || !am_generator) {
+ rprintf(FERROR, "invalid message %d:%d\n", tag, len);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ read_loop(fd, buf, len);
+ if (remove_source_files) {
+ decrement_active_files(IVAL(buf,0));
+ send_msg(MSG_SUCCESS, buf, len);
+ }
+ if (preserve_hard_links)
+ flist_ndx_push(&hlink_list, IVAL(buf,0));
+ break;
+ case MSG_SOCKERR:
+ if (!am_generator) {
+ rprintf(FERROR, "invalid message %d:%d\n", tag, len);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ close_multiplexing_out();
+ /* FALL THROUGH */
+ case MSG_INFO:
+ case MSG_ERROR:
+ case MSG_LOG:
+ while (len) {
+ n = len;
+ if (n >= sizeof buf)
+ n = sizeof buf - 1;
+ read_loop(fd, buf, n);
+ rwrite(tag, buf, n);
+ len -= n;
+ }
+ break;
+ default:
+ rprintf(FERROR, "unknown message %d:%d [%s]\n",
+ tag, len, who_am_i());
+ exit_cleanup(RERR_STREAMIO);
+ }
+
+ msg_fd_in = fd;
+}
+
+/* This is used by the generator to limit how many file transfers can
+ * be active at once when --remove-source-files is specified. Without
+ * this, sender-side deletions were mostly happening at the end. */
+void increment_active_files(int ndx, int itemizing, enum logcode code)
+{
+ /* TODO: tune these limits? */
+ while (active_filecnt >= (active_bytecnt >= 128*1024 ? 10 : 50)) {
+ if (hlink_list.head)
+ check_for_finished_hlinks(itemizing, code);
+ read_msg_fd();