approach
The Cro approach
At its heart, Cro is all about building up chains of Raku supplies (streams of asynchronous values) that process messages that arrive from the network and produce messages to be sent over the network.
Key roles
Messages are represented by Cro::Message. Concrete implementations include:
Cro::TCP::MessageCro::HTTP::RequestCro::HTTP::Response
Incoming connections are represented by Cro::Connection; implementations
include:
Cro::TCP::ServerConnectionCro::TLS::ServerConnection
A Cro::Source is a source of either messages or connections. For example,
Cro::TCP::Listener produces Cro::TCP::ServerConnection objects.
A Cro::Transform transforms one connection or message into another. For
example, Cro::HTTP::RequestParser will transform Cro::TCP::Messages into
Cro::HTTP::Requests, while a Cro::HTTP::ResponseSerializer transforms
Cro::HTTP::Responses into Cro::TCP::Messages. This means that in Cro,
HTTP applications are simply transforms from Cro::HTTP::Request into
Cro::HTTP::Response.
A Cro::Sink consumes messages. It doesn't produce anything. A sink comes at
the end of a message processing pipeline. A sink in a TCP server would consume
Cro::TCP::Messages and send them over the network.
Some messages or connections can be replied to with one or more messages. These
do the Cro::Replyable role. Anything that produces a replyable is also
responsible for providing something that can process the reply messages. This
"something" may either be a transform or a sink. Examples of replyables include
Cro::TCP::ServerConnection and Cro::TLS::ServerConnection, which give a
Cro::Sink replier that sends Cro::TCP::Message objects back to the client.
Composition
Cro components (sources, transforms, and sinks) can be put together to form pipelines. This process is called pipeline composition. Perhaps the simplest possible example is setting up an echo server:
class Echo does Cro::Transform {
method consumes() { Cro::TCP::Message }
method produces() { Cro::TCP::Message }
method reply(Supply $source) {
# We could actually just `return $source` here, but the identity
# supply is written out here to illustrate what a transform will
# often look like.
supply {
whenever $source -> $message {
emit $message;
}
}
}
}Which can then be composed into a service and started as follows:
my Cro::Service $echo-server = Cro.compose(
Cro::TCP::Listener.new(port => 8000),
Echo
);
$echo-server.start();Note that Cro.compose(...) only returns a Cro::Service in the case that
it has something that starts with a source and ends with a sink. So where did
the sink come from in this case? From the fact that a connection is replyable,
and so it provided the sink. It's also worth noting that a stream of connections
magically turned into a stream of messages. If the composer spots that something
producing connections is followed by something consuming messages, it will
pass the rest of the pipeline to a Cro::ConnectionManager instance, so the
processing of the remainder of the pipeline will be per connection.
An HTTP server example
Most Cro HTTP services will be assembled using high-level modules such as
Cro::HTTP::Router and Cro::HTTP::Server. However, it is possible to use
Cro.compose(...) to piece together a HTTP processing pipeline without these
conveniences. First, various components and message types are needed:
use Cro;
use Cro::HTTP::Request;
use Cro::HTTP::RequestParser;
use Cro::HTTP::Response;
use Cro::HTTP::ResponseSerializer;
use Cro::TCP;The HTTP application itself - a simple "Hello, world" - is a Cro::Transform
that turns a request into a response:
class HTTPHello does Cro::Transform {
method consumes() { Cro::HTTP::Request }
method produces() { Cro::HTTP::Response }
method transformer($request-stream) {
supply {
whenever $request-stream -> $request {
given Cro::HTTP::Response.new(:200status) {
.append-header('Content-type', 'text/html');
.set-body("<strong>Hello from Cro!</strong>");
.emit;
}
}
}
}
}These are composed into a service:
my Cro::Service $http-service = Cro.compose(
Cro::TCP::Listener.new( :host('localhost'), :port(8181) ),
Cro::HTTP::RequestParser.new,
HTTPHello,
Cro::HTTP::ResponseSerializer.new
);Which can then be used like this:
$http-service.start;
signal(SIGINT).tap: {
note "Shutting down...";
$http-service.stop;
exit;
}
sleep;Client pipelines
Clients, such as HTTP clients, are also expressed as pipelines. Unlike with server pipelines, where the application is at the center of the pipeline and the network at either end, a client pipeline has the network at the center and the application at either end.
The component at the center of a client pipeline will be a Cro::Connector,
which establishes a connection. A Cro::Connector is able to establish a
connection and produce a Cro::Transform that will send messages it consumes
using the connection and emit messages received from the network connection.
Pipelines featuring a connector must not have a Cro::Source nor a
Cro::Sink.
A minimal TCP client that connects, sends a message, and disconnects as soon as it has received something, could be expressed as:
my Cro::Connector $conn = Cro.compose(Cro::TCP::Connector);
my Supply $responses = $conn.establish(
host => 'localhost',
port => 4242,
supply {
emit Cro::TCP::Message.new( :data('hello'.encode('ascii')) )
}
);
react {
whenever $responses {
say .data;
done;
}
}The establish method on a client establishes a connection. It takes a single
positional argument with a Supply, which will be tapped to receive messages
to send; all named arguments will be passed along to the connect method,
which which makes a connection and returns a transform. The establish method
will return a Supply that the response messages will be emitted on.
More complex pipelines are possible. For example, a (not entirely convenient, but functional) HTTP client would look like:
my Cro::Connector $conn = Cro.compose(
Cro::HTTP::RequestSerializer,
Cro::TLS::Connector,
Cro::HTTP::ResponseParser
);
my $req = supply {
my Cro::HTTP::Request $req .= new(:method<GET>, :target</>);
$req.add-header('Host', 'www.raku.org');
emit $req;
}
react {
whenever $conn.establish($req, :host<www.raku.org>, :port(80)) {
say ~$response; # Dump headers
}
}