LCOV - code coverage report
Current view: top level - monetdb5/modules/atoms - streams.c (source / functions) Hit Total Coverage
Test: coverage.info Lines: 48 80 60.0 %
Date: 2021-10-13 02:24:04 Functions: 11 15 73.3 %

          Line data    Source code
       1             : /*
       2             :  * This Source Code Form is subject to the terms of the Mozilla Public
       3             :  * License, v. 2.0.  If a copy of the MPL was not distributed with this
       4             :  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
       5             :  *
       6             :  * Copyright 1997 - July 2008 CWI, August 2008 - 2021 MonetDB B.V.
       7             :  */
       8             : 
       9             : /*
      10             :  *  Niels Nes
      11             :  *  A simple interface to IO streams
      12             :  * All file IO is tunneled through the stream library, which guarantees
      13             :  * cross-platform capabilities.  Several protocols are provided, e.g. it
      14             :  * can be used to open 'non compressed, gzipped, bzip2ed' data files.  It
      15             :  * encapsulates the corresponding library managed in common/stream.
      16             :  */
      17             : 
      18             : #include "monetdb_config.h"
      19             : #include "streams.h"
      20             : #include "mal_exception.h"
      21             : 
      22           1 : static str mnstr_open_rstreamwrap(Stream *S, str *filename)
      23             : {
      24             :         stream *s;
      25             : 
      26           1 :         if ((s = open_rstream(*filename)) == NULL || mnstr_errnr(s)) {
      27           0 :                 if (s)
      28           0 :                         close_stream(s);
      29           0 :                 throw(IO, "streams.open", "could not open file '%s': %s",
      30             :                                 *filename, mnstr_peek_error(NULL));
      31             :         } else {
      32           1 :                 *(stream**)S = s;
      33             :         }
      34             : 
      35           1 :         return MAL_SUCCEED;
      36             : }
      37           1 : static str mnstr_open_wstreamwrap(Stream *S, str *filename)
      38             : {
      39             :         stream *s;
      40             : 
      41           1 :         if ((s = open_wstream(*filename)) == NULL || mnstr_errnr(s)) {
      42           0 :                 if (s)
      43           0 :                         close_stream(s);
      44           0 :                 throw(IO, "streams.open", "could not open file '%s': %s",
      45             :                                 *filename, mnstr_peek_error(NULL));
      46             :         } else {
      47           1 :                 *(stream**)S = s;
      48             :         }
      49             : 
      50           1 :         return MAL_SUCCEED;
      51             : }
      52             : 
      53           1 : static str mnstr_open_rastreamwrap(Stream *S, str *filename)
      54             : {
      55             :         stream *s;
      56             : 
      57           1 :         if ((s = open_rastream(*filename)) == NULL || mnstr_errnr(s)) {
      58           0 :                 if (s)
      59           0 :                         close_stream(s);
      60           0 :                 throw(IO, "streams.open", "could not open file '%s': %s",
      61             :                                 *filename, mnstr_peek_error(NULL));
      62             :         } else {
      63           1 :                 *(stream**)S = s;
      64             :         }
      65             : 
      66           1 :         return MAL_SUCCEED;
      67             : }
      68             : 
      69           1 : static str mnstr_open_wastreamwrap(Stream *S, str *filename)
      70             : {
      71             :         stream *s;
      72             : 
      73           1 :         if ((s = open_wastream(*filename)) == NULL || mnstr_errnr(s)) {
      74           0 :                 if (s)
      75           0 :                         close_stream(s);
      76           0 :                 throw(IO, "streams.open", "could not open file '%s': %s",
      77             :                                 *filename, mnstr_peek_error(NULL));
      78             :         } else {
      79           1 :                 *(stream**)S = s;
      80             :         }
      81             : 
      82           1 :         return MAL_SUCCEED;
      83             : }
      84             : 
      85             : static str
      86           1 : mnstr_write_stringwrap(void *ret, Stream *S, str *data)
      87             : {
      88           1 :         stream *s = *(stream **)S;
      89             :         (void)ret;
      90             : 
      91           1 :         if (mnstr_write(s, *data, 1, strlen(*data)) < 0)
      92           0 :                 throw(IO, "streams.writeStr", "failed to write string");
      93             : 
      94             :         return MAL_SUCCEED;
      95             : }
      96             : 
      97             : static str
      98           1 : mnstr_writeIntwrap(void *ret, Stream *S, int *data)
      99             : {
     100           1 :         stream *s = *(stream **)S;
     101             :         (void)ret;
     102             : 
     103           1 :         if (!mnstr_writeInt(s, *data))
     104           0 :                 throw(IO, "streams.writeInt", "failed to write int");
     105             : 
     106             :         return MAL_SUCCEED;
     107             : }
     108             : 
     109             : static str
     110           1 : mnstr_readIntwrap(int *ret, Stream *S)
     111             : {
     112           1 :         stream *s = *(stream **)S;
     113             : 
     114           1 :         if (mnstr_readInt(s, ret) != 1)
     115           0 :                 throw(IO, "streams.readInt", "failed to read int");
     116             : 
     117             :         return MAL_SUCCEED;
     118             : }
     119             : 
     120             : #define CHUNK (64 * 1024)
     121             : static str
     122           1 : mnstr_read_stringwrap(str *res, Stream *S)
     123             : {
     124           1 :         stream *s = *(stream **)S;
     125             :         ssize_t len = 0;
     126             :         size_t size = CHUNK + 1;
     127           1 :         char *buf = GDKmalloc(size), *start = buf, *tmp;
     128             : 
     129           1 :         if( buf == NULL)
     130           0 :                 throw(MAL,"mnstr_read_stringwrap", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     131           2 :         while ((len = mnstr_read(s, start, 1, CHUNK)) > 0) {
     132           1 :                 size += len;
     133           1 :                 tmp = GDKrealloc(buf, size);
     134           1 :                 if (tmp == NULL) {
     135           0 :                         GDKfree(buf);
     136           0 :                         throw(MAL,"mnstr_read_stringwrap", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     137             :                 }
     138             :                 buf = tmp;
     139           1 :                 start = buf + size - CHUNK - 1;
     140             : 
     141           1 :                 *start = '\0';
     142             :         }
     143           1 :         if (len < 0)
     144           0 :                 throw(IO, "streams.readStr", "failed to read string");
     145           1 :         start += len;
     146           1 :         *start = '\0';
     147           1 :         *res = buf;
     148             : 
     149           1 :         return MAL_SUCCEED;
     150             : }
     151             : 
     152             : static str
     153           1 : mnstr_flush_streamwrap(void *ret, Stream *S)
     154             : {
     155           1 :         stream *s = *(stream **)S;
     156             :         (void)ret;
     157             : 
     158           1 :         if (mnstr_flush(s, MNSTR_FLUSH_DATA))
     159           0 :                 throw(IO, "streams.flush", "failed to flush stream");
     160             : 
     161             :         return MAL_SUCCEED;
     162             : }
     163             : 
     164             : static str
     165           4 : mnstr_close_streamwrap(void *ret, Stream *S)
     166             : {
     167             :         (void)ret;
     168             : 
     169           4 :         close_stream(*(stream **)S);
     170             : 
     171           4 :         return MAL_SUCCEED;
     172             : }
     173             : 
     174             : static str
     175           0 : open_block_streamwrap(Stream *S, Stream *is)
     176             : {
     177           0 :         if ((*(stream **)S = block_stream(*(stream **)is)) == NULL)
     178           0 :                 throw(IO, "bstreams.open", "failed to open block stream");
     179             : 
     180             :         return MAL_SUCCEED;
     181             : }
     182             : 
     183             : static str
     184           0 : bstream_create_wrapwrap(Bstream *Bs, Stream *S, int *bufsize)
     185             : {
     186           0 :         if ((*(bstream **)Bs = bstream_create(*(stream **)S, (size_t)*bufsize)) == NULL)
     187           0 :                 throw(IO, "bstreams.create", "failed to create block stream");
     188             : 
     189             :         return MAL_SUCCEED;
     190             : }
     191             : 
     192             : static str
     193           0 : bstream_destroy_wrapwrap(void *ret, Bstream *BS)
     194             : {
     195             :         (void)ret;
     196             : 
     197           0 :         bstream_destroy(*(bstream **)BS);
     198             : 
     199           0 :         return MAL_SUCCEED;
     200             : }
     201             : 
     202             : static str
     203           0 : bstream_read_wrapwrap(int *res, Bstream *BS, int *size)
     204             : {
     205           0 :         *res = (int)bstream_read(*(bstream **)BS, (size_t)*size);
     206             : 
     207           0 :         return MAL_SUCCEED;
     208             : }
     209             : 
     210             : #include "mel.h"
     211             : mel_atom streams_init_atoms[] = {
     212             :  { .name="streams", .basetype="ptr", },
     213             :  { .name="bstream", .basetype="ptr", },  { .cmp=NULL }
     214             : };
     215             : mel_func streams_init_funcs[] = {
     216             :  command("streams", "openReadBytes", mnstr_open_rstreamwrap, true, "open a file stream for reading", args(1,2, arg("",streams),arg("filename",str))),
     217             :  command("streams", "openWriteBytes", mnstr_open_wstreamwrap, true, "open a file stream for writing", args(1,2, arg("",streams),arg("filename",str))),
     218             :  command("streams", "openRead", mnstr_open_rastreamwrap, true, "open ascii file stream for reading", args(1,2, arg("",streams),arg("filename",str))),
     219             :  command("streams", "openWrite", mnstr_open_wastreamwrap, true, "open ascii file stream for writing", args(1,2, arg("",streams),arg("filename",str))),
     220             :  command("streams", "blocked", open_block_streamwrap, true, "open a block based stream", args(1,2, arg("",streams),arg("s",streams))),
     221             :  command("streams", "writeStr", mnstr_write_stringwrap, true, "write data on the stream", args(1,3, arg("",void),arg("s",streams),arg("data",str))),
     222             :  command("streams", "writeInt", mnstr_writeIntwrap, true, "write data on the stream", args(1,3, arg("",void),arg("s",streams),arg("data",int))),
     223             :  command("streams", "readStr", mnstr_read_stringwrap, true, "read string data from the stream", args(1,2, arg("",str),arg("s",streams))),
     224             :  command("streams", "readInt", mnstr_readIntwrap, true, "read integer data from the stream", args(1,2, arg("",int),arg("s",streams))),
     225             :  command("streams", "flush", mnstr_flush_streamwrap, true, "flush the stream", args(0,1, arg("s",streams))),
     226             :  command("streams", "close", mnstr_close_streamwrap, true, "close and destroy the stream s", args(0,1, arg("s",streams))),
     227             :  command("streams", "create", bstream_create_wrapwrap, true, "create a buffered stream", args(1,3, arg("",bstream),arg("s",streams),arg("bufsize",int))),
     228             :  command("streams", "destroy", bstream_destroy_wrapwrap, true, "destroy bstream", args(0,1, arg("s",bstream))),
     229             :  command("streams", "read", bstream_read_wrapwrap, true, "read at least size bytes into the buffer of s", args(1,3, arg("",int),arg("s",bstream),arg("size",int))),
     230             :  { .imp=NULL }
     231             : };
     232             : #include "mal_import.h"
     233             : #ifdef _MSC_VER
     234             : #undef read
     235             : #pragma section(".CRT$XCU",read)
     236             : #endif
     237         259 : LIB_STARTUP_FUNC(init_streams_mal)
     238         259 : { mal_module("streams", streams_init_atoms, streams_init_funcs); }

Generated by: LCOV version 1.14