Line data    Source code

       1             : /*
       2             :  * SPDX-License-Identifier: MPL-2.0
       3             :  *
       4             :  * This Source Code Form is subject to the terms of the Mozilla Public
       5             :  * License, v. 2.0.  If a copy of the MPL was not distributed with this
       6             :  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
       7             :  *
       8             :  * Copyright 2024 MonetDB Foundation;
       9             :  * Copyright August 2008 - 2023 MonetDB B.V.;
      10             :  * Copyright 1997 - July 2008 CWI.
      11             :  */
      12             : 
      13             : #include "monetdb_config.h"
      14             : #include "stream.h"
      15             : #include "stream_internal.h"
      16             : 
      17             : 
      18             : 
      19             : 
      20             : /* ------------------------------------------------------------------ */
      21             : 
      22             : bstream *
      23      238813 : bstream_create(stream *s, size_t size)
      24             : {
      25      238813 :         bstream *b;
      26             : 
      27      238813 :         if (s == NULL)
      28             :                 return NULL;
      29      238813 :         if ((b = malloc(sizeof(*b))) == NULL)
      30             :                 return NULL;
      31      238813 :         *b = (bstream) {
      32             :                 .mode = size,
      33             :                 .s = s,
      34             :                 .eof = false,
      35             :         };
      36      238813 :         if (size == 0)
      37         359 :                 size = BUFSIZ;
      38      238813 :         b->buf = malloc(size + 1 + 1);
      39      238813 :         if (b->buf == NULL) {
      40           0 :                 free(b);
      41           0 :                 return NULL;
      42             :         }
      43      238813 :         b->size = size;
      44      238813 :         return b;
      45             : }
      46             : 
      47             : ssize_t
      48      917715 : bstream_read(bstream *s, size_t size)
      49             : {
      50      917715 :         ssize_t rd, rd1 = 0;
      51             : 
      52      917715 :         if (s == NULL)
      53             :                 return -1;
      54             : 
      55      917715 :         if (s->eof)
      56             :                 return 0;
      57             : 
      58      917713 :         assert(s->buf != NULL);
      59             : 
      60      917713 :         if (s->pos > 0) {
      61      338736 :                 if (s->pos < s->len) {
      62             :                         /* move all data and end of string marker */
      63        1571 :                         memmove(s->buf, s->buf + s->pos, s->len - s->pos + 1);
      64        1571 :                         s->len -= s->pos;
      65             :                 } else
      66      337165 :                         s->len = 0;
      67      338736 :                 s->pos = 0;
      68             :         }
      69             : 
      70      917713 :         if (s->len == s->size) {
      71          96 :                 size_t sz = size > 8192 ? 8192 : size;
      72          96 :                 char tmpbuf[8192];
      73             : 
      74             :                 /* before we realloc more space, see if there is a need */
      75          96 :                 if ((rd1 = s->s->read(s->s, tmpbuf, 1, sz)) == 0) {
      76           0 :                         s->eof = true;
      77           0 :                         return 0;
      78             :                 }
      79          96 :                 if (rd1 < 0)
      80             :                         return rd1;
      81          96 :                 char *p;
      82          96 :                 size_t ns = s->size + size;
      83          96 :                 if ((p = realloc(s->buf, ns + 1)) == NULL) {
      84             :                         return -1;
      85             :                 }
      86          96 :                 s->size = ns;
      87          96 :                 s->buf = p;
      88          96 :                 memcpy(s->buf + s->len, tmpbuf, rd1);
      89          96 :                 s->len += rd1;
      90          96 :                 size -= rd1;
      91          96 :                 if (size == 0)
      92             :                         return rd1;
      93             :         }
      94             : 
      95      917713 :         if (s->len + size > s->size)
      96      230703 :                 size = s->size - s->len;
      97             : 
      98      917713 :         rd = s->s->read(s->s, s->buf + s->len, 1, size);
      99             : 
     100      917709 :         if (rd < 0)
     101             :                 return rd;
     102             : 
     103      917567 :         if (rd == 0) {
     104      377117 :                 s->eof = true;
     105      377117 :                 return rd1;
     106             :         }
     107      540450 :         s->len += (size_t) rd;
     108      540450 :         s->buf[s->len] = 0;       /* fill in the spare with EOS */
     109      540450 :         return rd + rd1;
     110             : }
     111             : 
     112             : #ifdef _POSIX2_LINE_MAX
     113             : #define STREAM_LINE_MAX _POSIX2_LINE_MAX
     114             : #else
     115             : #define STREAM_LINE_MAX 2048
     116             : #endif
     117             : 
     118             : static ssize_t
     119           0 : bstream_readline(bstream *s)
     120             : {
     121           0 :         size_t size = STREAM_LINE_MAX;
     122           0 :         size_t rd;
     123             : 
     124           0 :         if (s->eof)
     125             :                 return 0;
     126             : 
     127           0 :         if (s->pos > 0 && s->len + size >= s->size) {
     128           0 :                 if (s->pos < s->len) {
     129             :                         /* move all data and end of string marker */
     130           0 :                         memmove(s->buf, s->buf + s->pos, s->len - s->pos + 1);
     131           0 :                         s->len -= s->pos;
     132             :                 } else
     133           0 :                         s->len = 0;
     134           0 :                 s->pos = 0;
     135             :         }
     136             : 
     137           0 :         assert(s->buf != NULL);
     138           0 :         if (s->len == s->size) {
     139           0 :                 char *p;
     140           0 :                 size_t ns = s->size + size + 8192;
     141           0 :                 if ((p = realloc(s->buf, ns + 1)) == NULL) {
     142             :                         return -1;
     143             :                 }
     144           0 :                 s->size = ns;
     145           0 :                 s->buf = p;
     146             :         }
     147             : 
     148           0 :         if (size > s->size - s->len)
     149             :                 size = s->size - s->len;
     150             : 
     151           0 :         if (fgets(s->buf + s->len, (int) size, s->s->stream_data.p) == NULL)
     152             :                 return -1;
     153             : 
     154           0 :         rd = strlen(s->buf + s->len);
     155             : 
     156           0 :         if (rd == 0) {
     157           0 :                 s->eof = true;
     158           0 :                 return 0;
     159             :         }
     160           0 :         s->len += rd;
     161           0 :         s->buf[s->len] = 0;       /* fill in the spare with EOS */
     162           0 :         return (ssize_t) rd;
     163             : }
     164             : 
     165             : 
     166             : ssize_t
     167      916154 : bstream_next(bstream *s)
     168             : {
     169      916154 :         if (s == NULL)
     170             :                 return -1;
     171      916154 :         if (s->mode > 0) {
     172      916154 :                 return bstream_read(s, s->mode);
     173           0 :         } else if (s->s->read == file_read) {
     174           0 :                 return bstream_readline(s);
     175             :         } else {
     176             :                 size_t sz = 0;
     177             :                 ssize_t rd;
     178             : 
     179           0 :                 while ((rd = bstream_read(s, 1)) == 1 &&
     180           0 :                        s->buf[s->pos + sz] != '\n') {
     181           0 :                         sz++;   /* sz += rd, but rd == 1 */
     182             :                 }
     183           0 :                 if (rd < 0)
     184             :                         return rd;
     185           0 :                 return (ssize_t) sz;
     186             :         }
     187             : }
     188             : 
     189             : void
     190      238812 : bstream_destroy(bstream *s)
     191             : {
     192      238812 :         if (s) {
     193      238812 :                 if (s->s) {
     194      238395 :                         s->s->close(s->s);
     195      238395 :                         s->s->destroy(s->s);
     196             :                 }
     197      238812 :                 if (s->buf)
     198      238812 :                         free(s->buf);
     199      238812 :                 free(s);
     200             :         }
     201      238812 : }
     202             :