view release on metacpan or  search on metacpan

Shared.xs  view on Meta::CPAN

  PREINIT:
    EXTRACT_HANDLE("Data::PubSub::Shared::Str", self);
  CODE:
    RETVAL = h->backing_fd;
  OUTPUT:
    RETVAL

void
DESTROY(self)
    SV *self
  CODE:
    if (!SvROK(self)) return;
    PubSubHandle *h = INT2PTR(PubSubHandle*, SvIV(SvRV(self)));
    if (!h) return;
    sv_setiv(SvRV(self), 0);
    pubsub_destroy(h);

SV *
publish(self, value)
    SV *self
    SV *value
  PREINIT:
    EXTRACT_HANDLE("Data::PubSub::Shared::Str", self);
  CODE:
    STRLEN len;
    SvGETMAGIC(value);   /* run get-magic/overload before reading the UTF8 flag */
    bool utf8 = SvUTF8(value) ? true : false;
    const char *str = SvPV_nomg(value, len);
    REEXTRACT_HANDLE("Data::PubSub::Shared::Str", self);
    int r = pubsub_str_publish(h, str, LEN32(len), utf8);
    if (r == -1) croak("publish: message too long (%u > %u)", (unsigned)len, h->msg_size);
    RETVAL = &PL_sv_yes;
  OUTPUT:
    RETVAL

UV
publish_multi(self, ...)
    SV *self
  PREINIT:
    EXTRACT_HANDLE("Data::PubSub::Shared::Str", self);
  CODE:
    RETVAL = 0;
    uint32_t count = items - 1;
    if (count > 8192) croak("publish_multi: too many values (%u > 8192)", count);
    if (count > 0) {
        /* Extract SV data BEFORE locking: SvPV(utf8) can run magic
         * (tied/overloaded stringification) that longjmps; doing it under
         * the process-shared mutex would abandon the lock and deadlock peers.
         * Newx+SAVEFREEPV so a die during extraction cannot leak args. */
        struct psm_arg { const char *str; STRLEN len; bool utf8; };
        struct psm_arg *args;
        Newx(args, count, struct psm_arg);
        SAVEFREEPV(args);
        for (uint32_t i = 0; i < count; i++) {
            SV *val = ST(i + 1);
            SvGETMAGIC(val);
            STRLEN len;
            const char *src = SvPV_nomg(val, len);
            /* Copy bytes into a private mortal SV NOW: a LATER element's
             * SvGETMAGIC (tied/overloaded) can run arbitrary Perl that
             * grows/frees THIS element's PV, dangling src before the locked
             * loop memcpys it. The mortal lives until FREETMPS after the loop. */
            SV *copy = sv_2mortal(newSVpvn(src, len));
            args[i].utf8 = SvUTF8(val) ? true : false;
            args[i].str = SvPVX_const(copy);
            args[i].len = len;
        }
        REEXTRACT_HANDLE("Data::PubSub::Shared::Str", self);
        pubsub_mutex_lock(h->hdr);
        for (uint32_t i = 0; i < count; i++) {
            int r = pubsub_str_publish_locked(h, args[i].str, LEN32(args[i].len), args[i].utf8);
            if (r == -1) {
                pubsub_mutex_unlock(h->hdr);
                croak("publish_multi: message too long (%u > %u)", (unsigned)args[i].len, h->msg_size);
            }
            RETVAL++;
        }
        pubsub_mutex_unlock(h->hdr);
        pubsub_wake_subscribers(h->hdr);
    }
  OUTPUT:
    RETVAL

void
publish_notify(self, value)
    SV *self
    SV *value
  PREINIT:
    EXTRACT_HANDLE("Data::PubSub::Shared::Str", self);
  CODE:
    STRLEN len;
    SvGETMAGIC(value);   /* run get-magic/overload before reading the UTF8 flag */
    bool utf8 = SvUTF8(value) ? true : false;
    const char *str = SvPV_nomg(value, len);
    REEXTRACT_HANDLE("Data::PubSub::Shared::Str", self);
    int r = pubsub_str_publish(h, str, LEN32(len), utf8);
    if (r == -1) croak("publish_notify: message too long (%u > %u)", (unsigned)len, h->msg_size);
    pubsub_notify(h);

SV *
subscribe(self)
    SV *self
  PREINIT:
    EXTRACT_HANDLE("Data::PubSub::Shared::Str", self);
  CODE:
    PubSubSub *sub = pubsub_subscribe(h, 0);
    if (!sub) croak("subscribe: out of memory");
    sub->userdata = (void *)newSVsv(self);
    MAKE_OBJ("Data::PubSub::Shared::Str::Sub", sub);
  OUTPUT:
    RETVAL

SV *
subscribe_all(self)
    SV *self
  PREINIT:
    EXTRACT_HANDLE("Data::PubSub::Shared::Str", self);
  CODE:
    PubSubSub *sub = pubsub_subscribe(h, 1);
    if (!sub) croak("subscribe_all: out of memory");
    sub->userdata = (void *)newSVsv(self);
    MAKE_OBJ("Data::PubSub::Shared::Str::Sub", sub);
  OUTPUT:
    RETVAL

UV
capacity(self)
    SV *self
  PREINIT:
    EXTRACT_HANDLE("Data::PubSub::Shared::Str", self);
  CODE: