contrib/serf/buckets/deflate_buckets.c

   1 /* Copyright 2002-2004 Justin Erenkrantz and Greg Stein
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <apr_strings.h>
  17
  18 #include <zlib.h>
  19
  20 /* This conditional isn't defined anywhere yet. */
  21 #ifdef HAVE_ZUTIL_H
  22 #include <zutil.h>
  23 #endif
  24
  25 #include "serf.h"
  26 #include "serf_bucket_util.h"
  27
  28 /* magic header */
  29 static char deflate_magic[2] = { '\037', '\213' };
  30 #define DEFLATE_MAGIC_SIZE 10
  31 #define DEFLATE_VERIFY_SIZE 8
  32 #define DEFLATE_BUFFER_SIZE 8096
  33
  34 static const int DEFLATE_WINDOW_SIZE = -15;
  35 static const int DEFLATE_MEMLEVEL = 9;
  36
  37 typedef struct {
  38     serf_bucket_t *stream;
  39     serf_bucket_t *inflate_stream;
  40
  41     int format;                 /* Are we 'deflate' or 'gzip'? */
  42
  43     enum {
  44         STATE_READING_HEADER,   /* reading the gzip header */
  45         STATE_HEADER,           /* read the gzip header */
  46         STATE_INIT,             /* init'ing zlib functions */
  47         STATE_INFLATE,          /* inflating the content now */
  48         STATE_READING_VERIFY,   /* reading the final gzip CRC */
  49         STATE_VERIFY,           /* verifying the final gzip CRC */
  50         STATE_FINISH,           /* clean up after reading body */
  51         STATE_DONE,             /* body is done; we'll return EOF here */
  52     } state;
  53
  54     z_stream zstream;
  55     char hdr_buffer[DEFLATE_MAGIC_SIZE];
  56     unsigned char buffer[DEFLATE_BUFFER_SIZE];
  57     unsigned long crc;
  58     int windowSize;
  59     int memLevel;
  60     int bufferSize;
  61
  62     /* How much of the chunk, or the terminator, do we have left to read? */
  63     apr_size_t stream_left;
  64
  65     /* How much are we supposed to read? */
  66     apr_size_t stream_size;
  67
  68     int stream_status; /* What was the last status we read? */
  69
  70 } deflate_context_t;
  71
  72 /* Inputs a string and returns a long.  */
  73 static unsigned long getLong(unsigned char *string)
  74 {
  75     return ((unsigned long)string[0])
  76           | (((unsigned long)string[1]) << 8)
  77           | (((unsigned long)string[2]) << 16)
  78           | (((unsigned long)string[3]) << 24);
  79 }
  80
  81 serf_bucket_t *serf_bucket_deflate_create(
  82     serf_bucket_t *stream,
  83     serf_bucket_alloc_t *allocator,
  84     int format)
  85 {
  86     deflate_context_t *ctx;
  87
  88     ctx = serf_bucket_mem_alloc(allocator, sizeof(*ctx));
  89     ctx->stream = stream;
  90     ctx->stream_status = APR_SUCCESS;
  91     ctx->inflate_stream = serf_bucket_aggregate_create(allocator);
  92     ctx->format = format;
  93     ctx->crc = 0;
  94     /* zstream must be NULL'd out. */
  95     memset(&ctx->zstream, 0, sizeof(ctx->zstream));
  96
  97     switch (ctx->format) {
  98         case SERF_DEFLATE_GZIP:
  99             ctx->state = STATE_READING_HEADER;
 100             break;
 101         case SERF_DEFLATE_DEFLATE:
 102             /* deflate doesn't have a header. */
 103             ctx->state = STATE_INIT;
 104             break;
 105         default:
 106             /* Not reachable */
 107             return NULL;
 108     }
 109
 110     /* Initial size of gzip header. */
 111     ctx->stream_left = ctx->stream_size = DEFLATE_MAGIC_SIZE;
 112
 113     ctx->windowSize = DEFLATE_WINDOW_SIZE;
 114     ctx->memLevel = DEFLATE_MEMLEVEL;
 115     ctx->bufferSize = DEFLATE_BUFFER_SIZE;
 116
 117     return serf_bucket_create(&serf_bucket_type_deflate, allocator, ctx);
 118 }
 119
 120 static void serf_deflate_destroy_and_data(serf_bucket_t *bucket)
 121 {
 122     deflate_context_t *ctx = bucket->data;
 123
 124     if (ctx->state > STATE_INIT &&
 125         ctx->state <= STATE_FINISH)
 126         inflateEnd(&ctx->zstream);
 127
 128     /* We may have appended inflate_stream into the stream bucket.
 129      * If so, avoid free'ing it twice.
 130      */
 131     if (ctx->inflate_stream) {
 132         serf_bucket_destroy(ctx->inflate_stream);
 133     }
 134     serf_bucket_destroy(ctx->stream);
 135
 136     serf_default_destroy_and_data(bucket);
 137 }
 138
 139 static apr_status_t serf_deflate_read(serf_bucket_t *bucket,
 140                                       apr_size_t requested,
 141                                       const char **data, apr_size_t *len)
 142 {
 143     deflate_context_t *ctx = bucket->data;
 144     unsigned long compCRC, compLen;
 145     apr_status_t status;
 146     const char *private_data;
 147     apr_size_t private_len;
 148     int zRC;
 149
 150     while (1) {
 151         switch (ctx->state) {
 152         case STATE_READING_HEADER:
 153         case STATE_READING_VERIFY:
 154             status = serf_bucket_read(ctx->stream, ctx->stream_left,
 155                                       &private_data, &private_len);
 156
 157             if (SERF_BUCKET_READ_ERROR(status)) {
 158                 return status;
 159             }
 160
 161             memcpy(ctx->hdr_buffer + (ctx->stream_size - ctx->stream_left),
 162                    private_data, private_len);
 163
 164             ctx->stream_left -= private_len;
 165
 166             if (ctx->stream_left == 0) {
 167                 ctx->state++;
 168                 if (APR_STATUS_IS_EAGAIN(status)) {
 169                     *len = 0;
 170                     return status;
 171                 }
 172             }
 173             else if (status) {
 174                 *len = 0;
 175                 return status;
 176             }
 177             break;
 178         case STATE_HEADER:
 179             if (ctx->hdr_buffer[0] != deflate_magic[0] ||
 180                 ctx->hdr_buffer[1] != deflate_magic[1]) {
 181                 return SERF_ERROR_DECOMPRESSION_FAILED;
 182             }
 183             if (ctx->hdr_buffer[3] != 0) {
 184                 return SERF_ERROR_DECOMPRESSION_FAILED;
 185             }
 186             ctx->state++;
 187             break;
 188         case STATE_VERIFY:
 189             /* Do the checksum computation. */
 190             compCRC = getLong((unsigned char*)ctx->hdr_buffer);
 191             if (ctx->crc != compCRC) {
 192                 return SERF_ERROR_DECOMPRESSION_FAILED;
 193             }
 194             compLen = getLong((unsigned char*)ctx->hdr_buffer + 4);
 195             if (ctx->zstream.total_out != compLen) {
 196                 return SERF_ERROR_DECOMPRESSION_FAILED;
 197             }
 198             ctx->state++;
 199             break;
 200         case STATE_INIT:
 201             zRC = inflateInit2(&ctx->zstream, ctx->windowSize);
 202             if (zRC != Z_OK) {
 203                 return SERF_ERROR_DECOMPRESSION_FAILED;
 204             }
 205             ctx->zstream.next_out = ctx->buffer;
 206             ctx->zstream.avail_out = ctx->bufferSize;
 207             ctx->state++;
 208             break;
 209         case STATE_FINISH:
 210             inflateEnd(&ctx->zstream);
 211             serf_bucket_aggregate_prepend(ctx->stream, ctx->inflate_stream);
 212             ctx->inflate_stream = 0;
 213             ctx->state++;
 214             break;
 215         case STATE_INFLATE:
 216             /* Do we have anything already uncompressed to read? */
 217             status = serf_bucket_read(ctx->inflate_stream, requested, data,
 218                                       len);
 219             if (SERF_BUCKET_READ_ERROR(status)) {
 220                 return status;
 221             }
 222             /* Hide EOF. */
 223             if (APR_STATUS_IS_EOF(status)) {
 224                 status = ctx->stream_status;
 225                 if (APR_STATUS_IS_EOF(status)) {
 226                     /* We've read all of the data from our stream, but we
 227                      * need to continue to iterate until we flush
 228                      * out the zlib buffer.
 229                      */
 230                     status = APR_SUCCESS;
 231                 }
 232             }
 233             if (*len != 0) {
 234                 return status;
 235             }
 236
 237             /* We tried; but we have nothing buffered. Fetch more. */
 238
 239             /* It is possible that we maxed out avail_out before
 240              * exhausting avail_in; therefore, continue using the
 241              * previous buffer.  Otherwise, fetch more data from
 242              * our stream bucket.
 243              */
 244             if (ctx->zstream.avail_in == 0) {
 245                 /* When we empty our inflated stream, we'll return this
 246                  * status - this allow us to eventually pass up EAGAINs.
 247                  */
 248                 ctx->stream_status = serf_bucket_read(ctx->stream,
 249                                                       ctx->bufferSize,
 250                                                       &private_data,
 251                                                       &private_len);
 252
 253                 if (SERF_BUCKET_READ_ERROR(ctx->stream_status)) {
 254                     return ctx->stream_status;
 255                 }
 256
 257                 if (!private_len && APR_STATUS_IS_EAGAIN(ctx->stream_status)) {
 258                     *len = 0;
 259                     status = ctx->stream_status;
 260                     ctx->stream_status = APR_SUCCESS;
 261                     return status;
 262                 }
 263
 264                 ctx->zstream.next_in = (unsigned char*)private_data;
 265                 ctx->zstream.avail_in = private_len;
 266             }
 267             zRC = Z_OK;
 268             while (ctx->zstream.avail_in != 0) {
 269                 /* We're full, clear out our buffer, reset, and return. */
 270                 if (ctx->zstream.avail_out == 0) {
 271                     serf_bucket_t *tmp;
 272                     ctx->zstream.next_out = ctx->buffer;
 273                     private_len = ctx->bufferSize - ctx->zstream.avail_out;
 274
 275                     ctx->crc = crc32(ctx->crc, (const Bytef *)ctx->buffer,
 276                                      private_len);
 277
 278                     /* FIXME: There probably needs to be a free func. */
 279                     tmp = SERF_BUCKET_SIMPLE_STRING_LEN((char *)ctx->buffer,
 280                                                         private_len,
 281                                                         bucket->allocator);
 282                     serf_bucket_aggregate_append(ctx->inflate_stream, tmp);
 283                     ctx->zstream.avail_out = ctx->bufferSize;
 284                     break;
 285                 }
 286                 zRC = inflate(&ctx->zstream, Z_NO_FLUSH);
 287
 288                 if (zRC == Z_STREAM_END) {
 289                     serf_bucket_t *tmp;
 290
 291                     private_len = ctx->bufferSize - ctx->zstream.avail_out;
 292                     ctx->crc = crc32(ctx->crc, (const Bytef *)ctx->buffer,
 293                                      private_len);
 294                     /* FIXME: There probably needs to be a free func. */
 295                     tmp = SERF_BUCKET_SIMPLE_STRING_LEN((char *)ctx->buffer,
 296                                                         private_len,
 297                                                         bucket->allocator);
 298                     serf_bucket_aggregate_append(ctx->inflate_stream, tmp);
 299
 300                     ctx->zstream.avail_out = ctx->bufferSize;
 301
 302                     /* Push back the remaining data to be read. */
 303                     tmp = serf_bucket_aggregate_create(bucket->allocator);
 304                     serf_bucket_aggregate_prepend(tmp, ctx->stream);
 305                     ctx->stream = tmp;
 306
 307                     /* We now need to take the remaining avail_in and
 308                      * throw it in ctx->stream so our next read picks it up.
 309                      */
 310                     tmp = SERF_BUCKET_SIMPLE_STRING_LEN(
 311                                         (const char*)ctx->zstream.next_in,
 312                                                      ctx->zstream.avail_in,
 313                                                      bucket->allocator);
 314                     serf_bucket_aggregate_prepend(ctx->stream, tmp);
 315
 316                     switch (ctx->format) {
 317                     case SERF_DEFLATE_GZIP:
 318                         ctx->stream_left = ctx->stream_size =
 319                             DEFLATE_VERIFY_SIZE;
 320                         ctx->state++;
 321                         break;
 322                     case SERF_DEFLATE_DEFLATE:
 323                         /* Deflate does not have a verify footer. */
 324                         ctx->state = STATE_FINISH;
 325                         break;
 326                     default:
 327                         /* Not reachable */
 328                         return APR_EGENERAL;
 329                     }
 330
 331                     break;
 332                 }
 333                 if (zRC != Z_OK) {
 334                     return SERF_ERROR_DECOMPRESSION_FAILED;
 335                 }
 336             }
 337             /* Okay, we've inflated.  Try to read. */
 338             status = serf_bucket_read(ctx->inflate_stream, requested, data,
 339                                       len);
 340             /* Hide EOF. */
 341             if (APR_STATUS_IS_EOF(status)) {
 342                 status = ctx->stream_status;
 343                 /* If our stream is finished too, return SUCCESS so
 344                  * we'll iterate one more time.
 345                  */
 346                 if (APR_STATUS_IS_EOF(status)) {
 347                     /* No more data to read from the stream, and everything
 348                        inflated. If all data was received correctly, state
 349                        should have been advanced to STATE_READING_VERIFY or
 350                        STATE_FINISH. If not, then the data was incomplete
 351                        and we have an error. */
 352                     if (ctx->state != STATE_INFLATE)
 353                         return APR_SUCCESS;
 354                     else
 355                         return SERF_ERROR_DECOMPRESSION_FAILED;
 356                 }
 357             }
 358             return status;
 359         case STATE_DONE:
 360             /* We're done inflating.  Use our finished buffer. */
 361             return serf_bucket_read(ctx->stream, requested, data, len);
 362         default:
 363             /* Not reachable */
 364             return APR_EGENERAL;
 365         }
 366     }
 367
 368     /* NOTREACHED */
 369 }
 370
 371 /* ### need to implement */
 372 #define serf_deflate_readline NULL
 373 #define serf_deflate_peek NULL
 374
 375 const serf_bucket_type_t serf_bucket_type_deflate = {
 376     "DEFLATE",
 377     serf_deflate_read,
 378     serf_deflate_readline,
 379     serf_default_read_iovec,
 380     serf_default_read_for_sendfile,
 381     serf_default_read_bucket,
 382     serf_deflate_peek,
 383     serf_deflate_destroy_and_data,
 384 };