IO::Stream(3pm)

💖 IO::Stream 💖

okay bestie, welcome to IO::Stream (3pm) — it's from the User Contributed Perl Documentation pile, and no cap it's a whole vibe. Let's get into it 🫶

---

✨ NAME ✨

IO::Stream — ease non-blocking I/O streams based on EV

🌸 VERSION 🌸

This document describes IO::Stream version v2.0.3

🦋 SYNOPSIS 🦋

use EV;
use IO::Stream;

IO::Stream->new({
    host        => 'google.com',
    port        => 80,
    cb          => \&client,
    wait_for    => SENT|EOF,
    in_buf_limit=> 102400,
    out_buf     => "GET / HTTP/1.0\nHost: google.com\n\n",
});

$EV::DIED = sub { warn $@; EV::unloop };
EV::loop;

sub client {
    my ($io, $e, $err) = @_;
    if ($err) {
        $io->close();
        die $err;
    }
    if ($e & SENT) {
        print "request sent, waiting for reply...\n";
    }
    if ($e & EOF) {
        print "server reply:\n", $io->{in_buf};
        $io->close();
        EV::unloop;         # ALL DONE
    }
}

💜 DESCRIPTION 💜

Okay so here's the tea ☕ — non-blocking, event-based, low-level I/O is so hard to get right. Like, code for it is usually error-prone and complex and honestly kinda a mess… and it ends up looking basically the same in every single app anyway. And it gets SO much worse the moment you need to alter the I/O stream in some way — route it through proxies, encrypt it, throw SSL at it, etc. 💀

This module exists to let you work with I/O streams on a higher level, using input/output buffers (just plain scalars, bestie) and high-level events like CONNECTED, SENT or EOF. But — and this is the slay part — it does not hide the low-level stuff from you. You can still work on the low level, no limitations, none. ✨

🧚 PLUGINS 🧚

The architecture of this module makes it easy to write plugins that alter the I/O stream in any way you want — route it through proxies, encrypt it, log it, whatever. The whole buffet. 🔥

Here are a few available plugins, and you can find more on CPAN: IO::Stream::Crypt::RC4, IO::Stream::Proxy::HTTPS, IO::Stream::MatrixSSL::Client, IO::Stream::MatrixSSL::Server.

If you're interested in writing your own plugin, go check the source for the "skeleton" plugins: IO::Stream::Noop and IO::Stream::NoopAlias. They're giving tutorial energy, fr fr. 👑

🎀 EXPORTS 🎀

This module doesn't export any functions/methods/variables, but it exports a LOT of constants. There are two groups of constants: events and errors (which you can import using the tags :Event and :Error). By default, all constants are exported.

Events:

RESOLVED CONNECTED IN OUT EOF SENT

Errors:

EINBUFLIMIT
ETORESOLVE ETOCONNECT ETOWRITE
EDNS EDNSNXDOMAIN EDNSNODATA
EREQINBUFLIMIT EREQINEOF

Errors are similar to $! — they're dualvars, meaning they have both textual and numeric values.

NOTE: Since v2.0.0 ETORESOLVE, EDNSNXDOMAIN and EDNSNODATA are not used anymore (EDNS is used instead), but they're still exported for compatibility. So like, they're still here, just chillin'. 🙈

🌈 OVERVIEW 🌈

You can create an IO::Stream object using any "stream" fh (file, TTY, UNIX socket, TCP socket, pipe, FIFO). Or, if you need a TCP socket, you can create an IO::Stream object using host+port instead of fh — in that case IO::Stream will do non-blocking host resolving, create the TCP socket and do a non-blocking connect. Iconic behavior, honestly. 💅

After you've created your IO::Stream object, it will handle read/write on this fh, and it will deliver only the high-level events you asked for into your callback, where you get to operate with in/out buffers instead of doing sysread()/syswrite() manually. Main character treatment.

There are no limitations on what you can do with the fh after you've created the IO::Stream object — you can even do sysread()/syswrite() (but like, there's no reason for you to do this anymore, bestie).

IMPORTANT! When you want to close this fh, you MUST use the $io->close() method for closing the fh instead of doing close($fh). This is because IO::Stream doesn't require you to keep the object returned by new(), and without a call to $io->close() the IO::Stream object will continue to exist and may receive/generate some events, which is not what you expect after closing the fh. Also, if you keep the object returned by IO::Stream->new() somewhere in your variables, you should either undef all such variables after you called $io->close(), or you should use Scalar::Util::weaken() on these variables after storing the IO::Stream object. (The same is applicable for all plugin objects too.) Don't skip this part, it's important!! 🚨

⭐ EVENTS ⭐

• RESOLVED — If you created the IO::Stream object using {host}+{port} instead of {fh}, this event will be generated after resolving {host}. The resolved IP address will be stored in {ip}.
• CONNECTED — If you created the IO::Stream object using {host}+{port} instead of {fh}, this event will be generated after connecting the socket to {ip}:{port}.
• IN — Generated after each successful read. IO::Stream may execute several sysread()s at once before generating the IN event, for optimization. The read data will be stored in {in_buf}, and the {in_bytes} counter will be incremented by the amount of bytes read.
• EOF — Generated only ONCE when EOF is reached (sysread() returns 0). Also will set {is_eof} to true.
• OUT — Generated when some data from {out_buf} was written. The written bytes are either removed from {out_buf} or just increment {out_pos} by the amount of bytes written (see the documentation about these fields below for more details). Also increments the {out_bytes} counter by the amount of bytes written.

Here, 'written' may be somewhat virtual — while {out_buf}/{out_pos} changes, the real data can still be sitting in plugin buffers (if you use plugins) and the real syswrite() may not have been called yet. To detect when all data is really written you should use the SENT event, not OUT. iykyk 🫶

• SENT — Generated when all data from {out_buf} was written. It's usual and safe to call $io->close() on the SENT event.

💫 TIMEOUTS 💫

IO::Stream has 30-second timeouts for connect and write; to timeout DNS resolve, use the default AnyEvent::DNS timeout. If you need to timeout other operations, you have to create your own timers using EV::timer().

Current version doesn't allow you to change these timeouts. Sorry bestie, them's the rules. 😭

🍓 SERVER 🍓

If you need to run a TCP/UNIX-server socket, then you should handle that socket manually. But you can create an IO::Stream object for an accept()'ed socket:

my ($host, $port) = ('0.0.0.0', 1234);
socket  my $srv_sock, AF_INET, SOCK_STREAM, 0;
setsockopt $srv_sock, SOL_SOCKET, SO_REUSEADDR, 1;
bind       $srv_sock, sockaddr_in($port, inet_aton($host));
listen     $srv_sock, SOMAXCONN;
fcntl      $srv_sock, F_SETFL, O_NONBLOCK;
$srv_w = EV::io($srv_sock, EV::READ, sub {
    if (accept my $sock, $srv_sock) {
        IO::Stream->new({
            fh          => $sock,
            cb          => \&server,
            wait_for    => IN,
        });
    }
    elsif ($! != EAGAIN) {
        die "accept: $!";
    }
});

💅 INTERFACE 💅

IO::Stream provides only three public methods: new(), write() and close(). new() will create a new object, close() will destroy it, and write() must be called when you want to modify (or just modified) the output buffer.

All other operations are done using IO::Stream object fields — for simplicity and performance reasons. Moreover, you can keep your own data in it too. There's a convention on field names, to avoid conflicts:

• /^_/ — Fields with names started with an underscore are for internal use by IO::Stream. You shouldn't touch them or create your own fields with such names. Hands off, that's private. 🙈
• /^[a-z]/ — Fields with names started with a lower-case letter are part of the IO::Stream public interface — you're allowed to read/write these fields, but you should not store incorrect values in these fields. Check "PUBLIC FIELDS" below for a description of available fields and their format.
• /^[A-Z]/ — You can store your own data in the IO::Stream object using field names started with an upper-case letter. IO::Stream will not touch these fields. That's your little corner. 🎀

When some event arises that you're waiting for, your callback will be called with 3 parameters: the IO::Stream object, the event mask, and the error (if any):

sub callback {
    my ($io, $e, $err) = @_;
}

🔥 METHODS 🔥

💖 new 💖

IO::Stream->new( \%opt );

Create and return an IO::Stream object. You may not keep the returned object — you will get it in your callback (in the first parameter) when some event interesting to you happens, and it will exist until you call the method close(). See OVERVIEW for more details.

Fields of %opt become fields of the created IO::Stream object. There are only a few fields required, but you can set any other fields too, and you can also set your custom fields (with names starting from an upper-case letter). Do your thing. ✨

The only required fields in %opt are {cb} and either {fh} or {host}+{port}. The {wait_for} field is also highly recommended to set when creating the object.

If {out_buf} will be set, then new() will automatically call write() after creating the object.

IO::Stream->new({
    fh          => \*STDIN,
    cb          => \&console,
    wait_for    => IN,
});

💖 write 💖

$io->write();
$io->write($data);

The method write() MUST be called after any modifications of the {out_buf} field, to ensure the data in {out_buf} will be written to {fh} as soon as it will be possible. Periodt. 💅

If {fh} is available for writing when calling write(), then it will write (may be partially) {out_buf} and may immediately call your callback function, delivering OUT|SENT events there. So, if you call write() from that callback (as it usually happens), keep in mind it may be called again while executing write(), and the object state may significantly change (it even may be close()'d) after it returns from write() into your callback. Heads up, bestie — this one's lowkey a plot twist. 😭

The write($data) is just a shortcut for:

$io->{out_buf} .= $data;
$io->write();

💖 close 💖

$io->close()

The method close() will close {fh} and destroy the IO::Stream object. See OVERVIEW for more details. 💜

✨ PUBLIC FIELDS ✨

okay so bestie, first the rules of the land: if a field is marked RO, that means it's read-only and you should not change it, periodt. Some fields have default values — they're shown after an equal sign. Some fields get modified on events. got it? slay.

• cb — the user callback</p> that gets called when any event listed in {wait_for} pops off or an error happens. Field {cb} should be either a CODE ref, or an object, or a class name. In the last two cases, a method named {method} will be called. Field {method} should be a string. 💅
• method = 'IO' — the method name that gets called when {cb} is an object or class name. Should be a string, bestie. no cap.
• wait_for — bitmask of events you're interested in. Can be changed at any time. For example:

$io->{wait_for} = RESOLVED|CONNECTED|IN|EOF|OUT|SENT;

When some data will be read from {fh}, {wait_for} must contain IN and/or EOF, or error EREQINEOF will be generated. So, it's better to always have IN and/or EOF in {wait_for}. If {wait_for} contain EOF and doesn't contain IN then {in_buf_limit} must be defined or error EREQINBUFLIMIT will be generated.

• fh RO — file handle for doing I/O. It's either provided by user to new(), or created by new() (when user provided {host}+{port} instead).
• host RO, port RO — if user doesn't provide {fh} to new(), he should provide {host} and {port} instead. This way new() will create new TCP socket in {fh} and resolve {host} and connect this {fh} to resolved {ip} and {port}. Both resolving and connecting happens in non-blocking way, and will result in delivering RESOLVED and CONNECTED events into user callback (if user {wait_for} these events).
• in_buf_limit = undef — used to avoid DoS attack when user doesn't handle IN events and want his callback called only on EOF event. Must be defined if user have EOF without IN in {wait_for}. Any value >0 will defined amount of bytes which can be read into {in_buf} before EOF happens. When size of {in_buf} become larger than {in_buf_limit}, error EINBUFLIMIT will be delivered to user callback. In this case user can either remove some data from {in_buf} to make it smaller than {in_buf_limit} or increase {in_buf_limit}, and continue reading data. NOT RECOMMENDED! Value 0 will switch off DoS protection, so there will be no limit on amount of data to read into {in_buf} until EOF happens.
• out_buf = q{} # modified on: OUT
• out_pos = undef # modified on: OUT Data from {out_buf} will be written to {fh}. If {out_pos} not defined, then data will be written from beginning of {out_buf}, and after successful write written bytes will be removed from beginning of {out_buf}. If {out_pos} defined, it should be >= 0. In this case data will be written from {out_pos} position in {out_buf}, and after successful write {out_pos} will be incremented by amount of bytes written. {out_buf} will not be changed!
• out_bytes = 0 # modified on: OUT — each successful write will increment {out_bytes} by amount of written bytes. You can change {out_bytes} in any way, but it should always be a number.
• in_buf = q{} # modified on: IN — each successful read will concatenate read bytes to {in_buf}. You can change {in_buf} in any way, but it should always be a string.
• in_bytes = 0 # modified on: IN — each successful read will increment {in_bytes} by amount of read bytes. You can change {in_bytes} in any way, but it should always be a number.
• ip RO = undef # modified on: RESOLVED — when you call new() with {host}+{port} instead of {fh} then IP address resolved from {host} will be stored in {ip}, and event RESOLVED will be generated.
• is_eof RO = undef # modified on: EOF — when EOF event happens {is_eof} will be set to true value. This allow you to detect is EOF already happens at any time, even if you doesn't have EOF in {wait_for}.
• plugin RO = {} — allow you to set list of plugins when creating object with new(), and later access these plugins. This field is somewhat special, because when you call new() you should set plugin to ARRAY ref, but in IO::Stream object {plugin} is HASH ref:

my $io = IO::Stream->new({
    host        => 'www.google.com',
    port        => 443,
    cb          => \&google,
    wait_for    => EOF,
    in_buf_limit=> 102400,
    out_buf     => "GET / HTTP/1.0\nHost: www.google.com\n\n",
    plugin      => [    # <------ it's ARRAY, but looks like HASH
        ssl         => IO::Stream::MatrixSSL::Client->new(),
        proxy       => IO::Stream::Proxy::HTTPS->new({
            host        => 'my.proxy.com',
            port        => 3218,
            user        => 'me',
            pass        => 'my pass',
        }),
    ],
    MyField1    => 'my data1',
    MyField2    => \%mydata2,
});

# access the "proxy" plugin:
$io->{plugin}{proxy};

This is because when calling new() it's important to keep plugins in order, but later it's easier to access them using names.

---

💀 DIAGNOSTICS 💀

Exceptions may be thrown only in new(). All other errors will be delivered to user's callback in last parameter.

• "usage: IO::Stream->new({ cb=>, wait_for=>, [fh=>, | host=>, port=>,] ... })" — you called new() with wrong parameters.
• "socket: %s", "fcntl: %s" — error happens while creating new socket. Usually this happens because you run out of file descriptors.
• "can't get file descriptor" — failed to get fileno() for your fh. Either fh doesn't open, or this fh type is not supported (directory handle), or fh is not file handle at all.
• "can't create second object for same fh" — you can't have more than one IO::Stream object for same fh. IO::Stream keep all objects created by new() until $io->close() will be called. Probably you've closed fh in some way without calling $io->close(), then new fh was created with same file descriptor number, and you've tried to create IO::Stream object using new fh.

---

👀 SEE ALSO 👀

AnyEvent::Handle — that's the whole see-also, bestie. go check it out. 🦋

---

🫶 SUPPORT 🫶

🐛 Bugs / Feature Requests 🐛

Please report any bugs or feature requests through the issue tracker at <https://github.com/powerman/perl-IO-Stream/issues>. You will be notified automatically of any progress on your issue.

📦 Source Code 📦

This is open source software. The code repository is available for public review and contribution under the terms of the license. Feel free to fork the repository and submit pull requests.

<https://github.com/powerman/perl-IO-Stream>

git clone https://github.com/powerman/perl-IO-Stream.git

🌈 Resources 🌈

• MetaCPAN Search — <https://metacpan.org/search?q=IO-Stream>
• CPAN Ratings — <http://cpanratings.perl.org/dist/IO-Stream>
• AnnoCPAN: Annotated CPAN documentation — <http://annocpan.org/dist/IO-Stream>
• CPAN Testers Matrix — <http://matrix.cpantesters.org/?dist=IO-Stream>
• CPANTS: A CPAN Testing Service (Kwalitee) — <http://cpants.cpanauthors.org/dist/IO-Stream>

---

👑 AUTHOR 👑

Alex Efros <powerman@cpan.org> — iconic behavior, honestly. 💖

---

📜 COPYRIGHT AND LICENSE 📜

This software is Copyright (c) 2008- by Alex Efros <powerman@cpan.org>.

This is free software, licensed under:

The MIT (X11) License

---

okay that's the whole tea, bestie. this page is from perl v5.36.0, dated 2022-08-28, and it's the IO::Stream(3pm) manpage. love you, bye! 💖✨🫶