Home | History | Annotate | Download | only in libcutils 
      1 /* libs/cutils/record_stream.c
      2 **
      3 ** Copyright 2006, The Android Open Source Project
      4 **
      5 ** Licensed under the Apache License, Version 2.0 (the "License");
      6 ** you may not use this file except in compliance with the License.
      7 ** You may obtain a copy of the License at
      8 **
      9 **     http://www.apache.org/licenses/LICENSE-2.0
     10 **
     11 ** Unless required by applicable law or agreed to in writing, software
     12 ** distributed under the License is distributed on an "AS IS" BASIS,
     13 ** WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     14 ** See the License for the specific language governing permissions and
     15 ** limitations under the License.
     16 */
     17 
     18 #include <stdlib.h>
     19 #include <unistd.h>
     20 #include <assert.h>
     21 #include <errno.h>
     22 #include <cutils/record_stream.h>
     23 #include <string.h>
     24 #include <stdint.h>
     25 #ifdef HAVE_WINSOCK
     26 #include <winsock2.h>   /* for ntohl */
     27 #else
     28 #include <netinet/in.h>
     29 #endif
     30 
     31 #define HEADER_SIZE 4
     32 
     33 struct RecordStream {
     34     int fd;
     35     size_t maxRecordLen;
     36 
     37     unsigned char *buffer;
     38 
     39     unsigned char *unconsumed;
     40     unsigned char *read_end;
     41     unsigned char *buffer_end;
     42 };
     43 
     44 
     45 extern RecordStream *record_stream_new(int fd, size_t maxRecordLen)
     46 {
     47     RecordStream *ret;
     48 
     49     assert (maxRecordLen <= 0xffff);
     50 
     51     ret = (RecordStream *)calloc(1, sizeof(RecordStream));
     52 
     53     ret->fd = fd;
     54     ret->maxRecordLen = maxRecordLen;
     55     ret->buffer = (unsigned char *)malloc (maxRecordLen + HEADER_SIZE);
     56 
     57     ret->unconsumed = ret->buffer;
     58     ret->read_end = ret->buffer;
     59     ret->buffer_end = ret->buffer + maxRecordLen + HEADER_SIZE;
     60 
     61     return ret;
     62 }
     63 
     64 
     65 extern void record_stream_free(RecordStream *rs)
     66 {
     67     free(rs->buffer);
     68     free(rs);
     69 }
     70 
     71 
     72 /* returns NULL; if there isn't a full record in the buffer */
     73 static unsigned char * getEndOfRecord (unsigned char *p_begin,
     74                                             unsigned char *p_end)
     75 {
     76     size_t len;
     77     unsigned char * p_ret;
     78 
     79     if (p_end < p_begin + HEADER_SIZE) {
     80         return NULL;
     81     }
     82 
     83     //First four bytes are length
     84     len = ntohl(*((uint32_t *)p_begin));
     85 
     86     p_ret = p_begin + HEADER_SIZE + len;
     87 
     88     if (p_end < p_ret) {
     89         return NULL;
     90     }
     91 
     92     return p_ret;
     93 }
     94 
     95 static void *getNextRecord (RecordStream *p_rs, size_t *p_outRecordLen)
     96 {
     97     unsigned char *record_start, *record_end;
     98 
     99     record_end = getEndOfRecord (p_rs->unconsumed, p_rs->read_end);
    100 
    101     if (record_end != NULL) {
    102         /* one full line in the buffer */
    103         record_start = p_rs->unconsumed + HEADER_SIZE;
    104         p_rs->unconsumed = record_end;
    105 
    106         *p_outRecordLen = record_end - record_start;
    107 
    108         return record_start;
    109     }
    110 
    111     return NULL;
    112 }
    113 
    114 /**
    115  * Reads the next record from stream fd
    116  * Records are prefixed by a 16-bit big endian length value
    117  * Records may not be larger than maxRecordLen
    118  *
    119  * Doesn't guard against EINTR
    120  *
    121  * p_outRecord and p_outRecordLen may not be NULL
    122  *
    123  * Return 0 on success, -1 on fail
    124  * Returns 0 with *p_outRecord set to NULL on end of stream
    125  * Returns -1 / errno = EAGAIN if it needs to read again
    126  */
    127 int record_stream_get_next (RecordStream *p_rs, void ** p_outRecord,
    128                                     size_t *p_outRecordLen)
    129 {
    130     void *ret;
    131 
    132     ssize_t countRead;
    133 
    134     /* is there one record already in the buffer? */
    135     ret = getNextRecord (p_rs, p_outRecordLen);
    136 
    137     if (ret != NULL) {
    138         *p_outRecord = ret;
    139         return 0;
    140     }
    141 
    142     // if the buffer is full and we don't have a full record
    143     if (p_rs->unconsumed == p_rs->buffer
    144         && p_rs->read_end == p_rs->buffer_end
    145     ) {
    146         // this should never happen
    147         //LOGE("max record length exceeded\n");
    148         assert (0);
    149         errno = EFBIG;
    150         return -1;
    151     }
    152 
    153     if (p_rs->unconsumed != p_rs->buffer) {
    154         // move remainder to the beginning of the buffer
    155         size_t toMove;
    156 
    157         toMove = p_rs->read_end - p_rs->unconsumed;
    158         if (toMove) {
    159             memmove(p_rs->buffer, p_rs->unconsumed, toMove);
    160         }
    161 
    162         p_rs->read_end = p_rs->buffer + toMove;
    163         p_rs->unconsumed = p_rs->buffer;
    164     }
    165 
    166     countRead = read (p_rs->fd, p_rs->read_end, p_rs->buffer_end - p_rs->read_end);
    167 
    168     if (countRead <= 0) {
    169         /* note: end-of-stream drops through here too */
    170         *p_outRecord = NULL;
    171         return countRead;
    172     }
    173 
    174     p_rs->read_end += countRead;
    175 
    176     ret = getNextRecord (p_rs, p_outRecordLen);
    177 
    178     if (ret == NULL) {
    179         /* not enough of a buffer to for a whole command */
    180         errno = EAGAIN;
    181         return -1;
    182     }
    183 
    184     *p_outRecord = ret;
    185     return 0;
    186 }
    187