Managing Multiple Streams of Input (Perl Cookbook, 2nd Edition)

## 17.19. Managing Multiple Streams of Input

### 17.19.1. Problem

The next input to your program could be coming from any number of filehandles, but you don't know which. You've tried using `select( )`, but the need to then do unbuffered I/O is more than you can deal with (and it's making your code very difficult to follow).

### 17.19.2. Solution

Use the IO::Multiplex module from CPAN. It calls a `mux_input( )` function when input is received over a socket, and handles input and output buffering for you:

>     use IO::Multiplex;
>     $mux = IO::Multiplex->new( );
>     $mux->add($FH1);
>     $mux->add($FH2); # ... and so on for all the filehandles to manage
>     $mux->set_callback_object(_ _PACKAGE_ _);  # or an object
>     $mux->Loop( );

>     sub mux_input {
>       my ($package, $mux, $fh, $input) = @_;
>       # $input is ref to the filehandle's input buffer
>       # ...
>     }

### 17.19.3. Discussion

Although you can use `select` to manage input coming at you from multiple directions, there are many tricks and traps. For example, you can't use `<>` to read a line of input, because you never know whether the client has sent a full line yet (or will ever finish sending a line). You can't `print` to a socket without the risk of the output buffer being full and your process blocking. You need to use non-blocking I/O and maintain your own buffers, and, consequently, life rapidly becomes unmanageably complex.

Fortunately, we have a way of hiding complexity: modules. The IO::Multiplex module from CPAN takes care of non-blocking I/O and `select` for you. You tell it which filehandles to watch, and it tells you when new data arrives. You can even `print` to the filehandles, and it'll buffer and non-blockingly output it. An IO::Multiplex object manages a pool of filehandles.

Use the `add` method to tell IO::Multiplex to manage a filehandle. This enables non-blocking I/O and disables the stdio buffering. When IO::Multiplex receives data on one of its managed filehandles, it calls a `mux_input` method on an object or class of your choosing. Specify where `mux_input` is by passing a package name (if your callback is a class method) or object value (if your callback is an object method) to the IO::Multiplex `set_callback_object` method. In the example in the Solution, we pass in the current package name so that IO::Multiplex will call the current package's `mux_input` method.

Your `mux_input` callback is called with four parameters: the object or package name that you gave to `set_callback_object`, the IO::Multiplex object that dispatched the callback, the filehandle from which data was received, and a reference to the input buffer. The callback should delete data from the buffer once it has been processed. For example, to process line by line:

>     sub mux_input {
>       my ($obj, $mux, $fh, $buffer) = @_;
>       my ($line) = $$buffer =~ s{^(.*)\n}{  } or return;
>       # ...
>     }

The IO::Multiplex module also takes care of `accept` ing incoming connections on server sockets. Once you have a socket bound and listening (see Recipe 17.2), pass it to the `listen` method of an IO::Multiplex object:

>     use IO::Socket;
>     $server = IO::Socket::INET->new(LocalPort => $PORT, Listen => 10)
>       or die $@;
>     $mux->listen($server);

When new incoming connections are accepted, the `mux_connection` callback is called. There are other callbacks, such as for full and partial closure of a filehandle, timeouts, and so on. For a full list of the methods you can use to control an IO::Multiplex object and a full list of the callbacks, see the IO::Multiplex documentation.

Example 17-7 is a rudimentary chat server that uses IO::Multiplex. It listens on port 6901 of the local host address and implements a very rudimentary chat protocol. Every client (see Example 17-8) has a "name," which they can change by sending a line that looks like `/nick newname`. Every other incoming line of text is sent out to all connected machines, prefaced with the name of the client that sent it.

To test this out, run the server in one window, then start a few clients in other windows. Type something into one and see what appears in the others.

#### Example 17-7. chatserver

>       #!/usr/bin/perl -w
>       # chatserver - very simple chat server
>       use IO::Multiplex;
>       use IO::Socket;
>       use strict;
>       my %Name;
>       my $Server = IO::Socket::INET->new(LocalAddr => "localhost:6901",
>                                          Listen   => 10, Reuse => 1,
>                                          Proto    => 'tcp') or die $@;
>       my $Mux = IO::Multiplex->new( );
>       my $Person_Counter = 1;
>       $Mux->listen($Server);
>       $Mux->set_callback_object(_ _PACKAGE_ _);
>       $Mux->loop( );
>       exit;
>       sub mux_connection {
>         my ($package, $mux, $fh) = @_;
>         $Name{$fh} = [ $fh, "Person " . $Person_Counter++ ];
>       }
>       sub mux_eof {
>         my ($package, $mux, $fh) = @_;
>         delete $Name{$fh};
>       }
>       sub mux_input {
>         my ($package, $mux, $fh, $input) = @_;
>         my $line;
>         my $name;
>         $$input =~ s{^(.*)\n+}{  } or return;
>         $line = $1;
>         if ($line =~ m{^/nick\s+(\S+)\s*}) {
>           my $oldname = $Name{$fh};
>           $Name{$fh} = [ $fh, $1 ];
>           $line = "$oldname->[1] is now known as $1";
>         } else {
>           $line = "<$Name{$fh}[1]> $line";
>         }
>         foreach my $conn_struct (values %Name) {
>           my $conn = $conn_struct->[0];
>           $conn->print("$line\n");
>         }
>       }

#### Example 17-8. chatclient

>       #!/usr/bin/perl -w
>       # chatclient - client for the chat server
>       use IO::Multiplex;
>       use IO::Socket;
>       use strict;
>       my $sock = IO::Socket::INET->new(PeerAddr => "localhost:6901",
>                                        Proto    => "tcp") or die $@;
>       my $Mux = IO::Multiplex->new( );
>       $Mux->add($sock);
>       $Mux->add(*STDIN);
>       $Mux->set_callback_object(_ _PACKAGE_ _);
>       $Mux->loop( );
>       exit;
>       sub mux_input {
>         my ($package, $mux, $fh, $input) = @_;
>         my $line;
>         $line = $$input;
>         $$input = "";
>         if (fileno($fh) =  = fileno(STDIN)) {
>           print $sock $line;
>         } else {
>           print $line;
>         }
>       }

### 17.19.4. See Also

The documentation for the CPAN module IO::Multiplex; Recipe 17.1, Recipe 17.2, Recipe 17.20, and Recipe 17.21

Copyright © 2003 O'Reilly & Associates. All rights reserved.