+ if (nb != 0 || flush_pending) {
+ /* deflate the data starting at offset */
+ int flush = Z_NO_FLUSH;
+ tx_strm.avail_in = 0;
+ tx_strm.avail_out = 0;
+ do {
+ if (tx_strm.avail_in == 0 && nb != 0) {
+ /* give it some more input */
+ n = MIN(nb, CHUNK_SIZE);
+ tx_strm.next_in = (Bytef *)
+ map_ptr(buf, offset, n);
+ tx_strm.avail_in = n;
+ nb -= n;
+ offset += n;
+ }
+ if (tx_strm.avail_out == 0) {
+ tx_strm.next_out = (Bytef *)(obuf + 2);
+ tx_strm.avail_out = MAX_DATA_COUNT;
+ if (flush != Z_NO_FLUSH) {
+ /*
+ * We left the last 4 bytes in the
+ * buffer, in case they are the
+ * last 4. Move them to the front.
+ */
+ memcpy(tx_strm.next_out,
+ obuf+MAX_DATA_COUNT-2, 4);
+ tx_strm.next_out += 4;
+ tx_strm.avail_out -= 4;
+ }
+ }
+ if (nb == 0 && token != -2)
+ flush = Z_SYNC_FLUSH;
+ r = deflate(&tx_strm, flush);
+ if (r != Z_OK) {
+ rprintf(FERROR, "deflate returned %d\n", r);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ if (nb == 0 || tx_strm.avail_out == 0) {
+ n = MAX_DATA_COUNT - tx_strm.avail_out;
+ if (flush != Z_NO_FLUSH) {
+ /*
+ * We have to trim off the last 4
+ * bytes of output when flushing
+ * (they are just 0, 0, ff, ff).
+ */
+ n -= 4;
+ }
+ if (n > 0) {
+ obuf[0] = DEFLATED_DATA + (n >> 8);
+ obuf[1] = n;
+ write_buf(f, obuf, n+2);
+ }
+ }
+ } while (nb != 0 || tx_strm.avail_out == 0);
+ flush_pending = token == -2;
+ }
+
+ if (token == -1) {
+ /* end of file - clean up */
+ write_byte(f, END_FLAG);
+ } else if (token != -2) {
+ /* Add the data in the current block to the compressor's
+ * history and hash table. */
+ do {
+ /* Break up long sections in the same way that
+ * see_deflate_token() does. */
+ int32 n1 = toklen > 0xffff ? 0xffff : toklen;
+ toklen -= n1;
+ tx_strm.next_in = (Bytef *)map_ptr(buf, offset, n1);
+ tx_strm.avail_in = n1;
+ tx_strm.next_out = (Bytef *) obuf;
+ tx_strm.avail_out = AVAIL_OUT_SIZE(CHUNK_SIZE);
+ r = deflate(&tx_strm, Z_INSERT_ONLY);
+ if (r != Z_OK || tx_strm.avail_in != 0) {
+ rprintf(FERROR, "deflate on token returned %d (%d bytes left)\n",
+ r, tx_strm.avail_in);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ } while (toklen > 0);
+ }
+}
+
+/* tells us what the receiver is in the middle of doing */
+static enum { r_init, r_idle, r_running, r_inflating, r_inflated } recv_state;
+
+/* for inflating stuff */
+static z_stream rx_strm;
+static char *cbuf;
+static char *dbuf;
+
+/* for decoding runs of tokens */
+static int32 rx_token;
+static int32 rx_run;
+
+/* Receive a deflated token and inflate it */
+static int32 recv_deflated_token(int f, char **data)
+{
+ static int init_done;
+ static int32 saved_flag;
+ int32 n, flag;
+ int r;
+
+ for (;;) {
+ switch (recv_state) {
+ case r_init:
+ if (!init_done) {
+ rx_strm.next_out = NULL;
+ rx_strm.zalloc = NULL;
+ rx_strm.zfree = NULL;
+ if (inflateInit2(&rx_strm, -15) != Z_OK) {
+ rprintf(FERROR, "inflate init failed\n");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ if (!(cbuf = new_array(char, MAX_DATA_COUNT))
+ || !(dbuf = new_array(char, AVAIL_OUT_SIZE(CHUNK_SIZE))))
+ out_of_memory("recv_deflated_token");
+ init_done = 1;
+ } else {
+ inflateReset(&rx_strm);
+ }
+ recv_state = r_idle;
+ rx_token = 0;
+ break;
+
+ case r_idle:
+ case r_inflated:
+ if (saved_flag) {
+ flag = saved_flag & 0xff;
+ saved_flag = 0;
+ } else
+ flag = read_byte(f);
+ if ((flag & 0xC0) == DEFLATED_DATA) {
+ n = ((flag & 0x3f) << 8) + read_byte(f);
+ read_buf(f, cbuf, n);
+ rx_strm.next_in = (Bytef *)cbuf;
+ rx_strm.avail_in = n;
+ recv_state = r_inflating;
+ break;
+ }
+ if (recv_state == r_inflated) {
+ /* check previous inflated stuff ended correctly */
+ rx_strm.avail_in = 0;
+ rx_strm.next_out = (Bytef *)dbuf;
+ rx_strm.avail_out = AVAIL_OUT_SIZE(CHUNK_SIZE);
+ r = inflate(&rx_strm, Z_SYNC_FLUSH);
+ n = AVAIL_OUT_SIZE(CHUNK_SIZE) - rx_strm.avail_out;
+ /*
+ * Z_BUF_ERROR just means no progress was
+ * made, i.e. the decompressor didn't have
+ * any pending output for us.
+ */
+ if (r != Z_OK && r != Z_BUF_ERROR) {
+ rprintf(FERROR, "inflate flush returned %d (%d bytes)\n",
+ r, n);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ if (n != 0 && r != Z_BUF_ERROR) {
+ /* have to return some more data and
+ save the flag for later. */
+ saved_flag = flag + 0x10000;
+ *data = dbuf;
+ return n;
+ }
+ /*
+ * At this point the decompressor should
+ * be expecting to see the 0, 0, ff, ff bytes.
+ */
+ if (!inflateSyncPoint(&rx_strm)) {
+ rprintf(FERROR, "decompressor lost sync!\n");
+ exit_cleanup(RERR_STREAMIO);
+ }
+ rx_strm.avail_in = 4;
+ rx_strm.next_in = (Bytef *)cbuf;
+ cbuf[0] = cbuf[1] = 0;
+ cbuf[2] = cbuf[3] = 0xff;
+ inflate(&rx_strm, Z_SYNC_FLUSH);
+ recv_state = r_idle;
+ }
+ if (flag == END_FLAG) {
+ /* that's all folks */
+ recv_state = r_init;
+ return 0;
+ }
+
+ /* here we have a token of some kind */
+ if (flag & TOKEN_REL) {
+ rx_token += flag & 0x3f;
+ flag >>= 6;
+ } else
+ rx_token = read_int(f);
+ if (flag & 1) {
+ rx_run = read_byte(f);
+ rx_run += read_byte(f) << 8;
+ recv_state = r_running;
+ }
+ return -1 - rx_token;
+
+ case r_inflating:
+ rx_strm.next_out = (Bytef *)dbuf;
+ rx_strm.avail_out = AVAIL_OUT_SIZE(CHUNK_SIZE);
+ r = inflate(&rx_strm, Z_NO_FLUSH);
+ n = AVAIL_OUT_SIZE(CHUNK_SIZE) - rx_strm.avail_out;
+ if (r != Z_OK) {
+ rprintf(FERROR, "inflate returned %d (%d bytes)\n", r, n);
+ exit_cleanup(RERR_STREAMIO);
+ }
+ if (rx_strm.avail_in == 0)
+ recv_state = r_inflated;
+ if (n != 0) {
+ *data = dbuf;
+ return n;
+ }
+ break;
+
+ case r_running:
+ ++rx_token;
+ if (--rx_run == 0)
+ recv_state = r_idle;
+ return -1 - rx_token;
+ }
+ }
+}