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