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             : /* callback stream
      21             :  *
      22             :  * read-only stream which calls a user-provided callback function in
      23             :  * order to get more data to be returned to the reader */
      24             : 
      25             : struct cbstream {
      26             :         void *private;
      27             :         void (*destroy)(void *);
      28             :         void (*close)(void *);
      29             :         ssize_t (*read)(void *, void *, size_t, size_t);
      30             :         ssize_t (*write)(void *, const void *, size_t, size_t);
      31             : };
      32             : 
      33             : static void
      34           4 : cb_destroy(stream *s)
      35             : {
      36           4 :         struct cbstream *cb = s->stream_data.p;
      37             : 
      38           4 :         if (cb->destroy)
      39           4 :                 cb->destroy(cb->private);
      40           4 :         free(cb);
      41           4 :         s->stream_data.p = NULL;
      42           4 :         destroy_stream(s);
      43           4 : }
      44             : 
      45             : static void
      46           0 : cb_close(stream *s)
      47             : {
      48           0 :         struct cbstream *cb = s->stream_data.p;
      49             : 
      50           0 :         if (cb->close)
      51           0 :                 cb->close(cb->private);
      52           0 : }
      53             : 
      54             : static ssize_t
      55           0 : cb_read(stream *restrict s, void *restrict buf, size_t elmsize, size_t cnt)
      56             : {
      57           0 :         struct cbstream *cb = s->stream_data.p;
      58             : 
      59           0 :         ssize_t ret = cb->read(cb->private, buf, elmsize, cnt);
      60           0 :         s->eof |= ret == 0;
      61           0 :         return ret;
      62             : }
      63             : 
      64             : static ssize_t
      65         372 : cb_write(stream *restrict s, const void *restrict buf, size_t elmsize, size_t cnt)
      66             : {
      67         372 :         struct cbstream *cb = s->stream_data.p;
      68             : 
      69         372 :         return cb->write(cb->private, buf, elmsize, cnt);
      70             : }
      71             : 
      72             : stream *
      73           4 : callback_stream(void *restrict private,
      74             :                 ssize_t (*read)(void *restrict private, void *restrict buf, size_t elmsize, size_t cnt),
      75             :                 ssize_t (*write)(void *restrict private, const void *restrict buf, size_t elmsize, size_t cnt),
      76             :                 void (*close)(void *private),
      77             :                 void (*destroy)(void *private),
      78             :                 const char *restrict name)
      79             : {
      80           4 :         stream *s;
      81           4 :         struct cbstream *cb;
      82             : 
      83           4 :         if ((write && read) || (!write && !read))
      84             :                 return NULL;
      85             : 
      86           4 :         s = create_stream(name);
      87           4 :         if (s == NULL)
      88             :                 return NULL;
      89           4 :         cb = malloc(sizeof(struct cbstream));
      90           4 :         if (cb == NULL) {
      91           0 :                 destroy_stream(s);
      92           0 :                 mnstr_set_open_error(name, errno, NULL);
      93           0 :                 return NULL;
      94             :         }
      95           4 :         *cb = (struct cbstream) {
      96             :                 .private = private,
      97             :                 .destroy = destroy,
      98             :                 .read = read,
      99             :                 .write = write,
     100             :                 .close = close,
     101             :         };
     102           4 :         s->readonly = (read != NULL);
     103           4 :         s->stream_data.p = cb;
     104           4 :         s->read = read ? cb_read : NULL;
     105           4 :         s->write = write ? cb_write : NULL;
     106           4 :         s->destroy = cb_destroy;
     107           4 :         s->close = cb_close;
     108           4 :         return s;
     109             : }