ggl · February 18, 2022 09:34
diff --git a/pubsub.pl b/pubsub.pl
 #!/usr/bin/env perl
 # ABSTRACT: A simple message broker using Mojolicious WebSockets
 # USAGE: ./pubsub.pl daemon
 #
 # This is my remix of Doug Bell's event broker. It publishes numbers
 # on a topic, subscribes to that topic and appends the numbers to the
 # document body. The original source can be found here:
 #
 # https://gist.github.com/preaction/2078d33d87b126621e45
 #
 # Copyright 2015 Doug Bell (<[email protected]>)
 # 
 # This is free software; you can redistribute it and/or modify it under
 # the same terms as the Perl 5 programming language system itself.

 use Mojolicious::Lite;
 use Data::Dumper;
 use Scalar::Util qw(refaddr);

 my %topics;

 helper add_topic_subscriber => sub {
 	my ($self, $topic) = @_;
 	$topics{$topic}{refaddr $self} = $self;
 	return;
 };

 helper remove_topic_subscriber => sub {
 	my ($self, $topic) = @_;
 	delete $topics{$topic}{refaddr $self};
 	return;
 };

 helper publish_topic_message => sub {
 	my ($self, $topic, $msg) = @_;
 	$_->send($msg) for values %{ $topics{$topic} };
 	return;
 };

 # Template with browser-side code
 get '/' => 'index';

 # WebSocket echo service
 websocket '/echo' => sub {
 	my $c = shift;
 	# Opened
 	$c->app->log->debug('WebSocket opened');
 	# Increase inactivity timeout for connection a bit
 	$c->inactivity_timeout(120);
 	# Incoming message
 	$c->on(message => sub {
 		my ($c, $msg) = @_;
 		$c->send("echo: $msg");
 	});
 	# Closed
 	$c->on(finish => sub {
 		my ($c, $code, $reason) = @_;
 		$c->app->log->debug("WebSocket closed with status $code");
 	});
 };

 websocket '/pub/*topic' => sub {
 	my $c = shift;
 	$c->inactivity_timeout(120);
 	my $topic = $c->stash('topic');
 	$c->app->log->debug($c->tx->remote_address.' publish to '.$topic);
 	$c->on(message => sub {
 		my ($c, $msg) = @_;
 		$c->publish_topic_message($topic, $msg);
 	} );
 };

 websocket '/sub/*topic' => sub {
 	my $c = shift;
 	$c->inactivity_timeout(120);
 	my $topic = $c->stash('topic');
 	$c->add_topic_subscriber($topic);
 	$c->app->log->debug($c->tx->remote_address.' subscribed to '.$topic);
 	$c->on(finish => sub {
 		my $c = shift;
 		$c->remove_topic_subscriber($topic);
 	} );
 };

 app->secrets([join('', map(("a".."z","A".."Z","0".."9")[rand 62], 0..20))]);
 app->start;
 __DATA__

 @@ index.html.ep
 <!DOCTYPE html>
 <html>
 	<head><title>Echo</title></head>
 	<body>
 		<script>
 			var count = 1;
 			var pub = new WebSocket('<%= $self->url_for('pub/blah')->to_abs->scheme('ws') %>');
 			var sub = new WebSocket('<%= $self->url_for('sub/blah')->to_abs->scheme('ws') %>');
 			sub.onmessage = function (event) {
 				document.body.appendChild(document.createTextNode(event.data+' '));
 			};
 			pub.onopen = function (event) {
 				window.setInterval(function () { pub.send(count++) }, 1000);
 			};
 		</script>
 	</body>
 </html>
	#!/usr/bin/env perl
	# ABSTRACT: A simple message broker using Mojolicious WebSockets
	# USAGE: ./pubsub.pl daemon
	#
	# This is my remix of Doug Bell's event broker. It publishes numbers
	# on a topic, subscribes to that topic and appends the numbers to the
	# document body. The original source can be found here:
	#
	# https://gist.github.com/preaction/2078d33d87b126621e45
	#
	# Copyright 2015 Doug Bell (<[email protected]>)
	#
	# This is free software; you can redistribute it and/or modify it under
	# the same terms as the Perl 5 programming language system itself.

	use Mojolicious::Lite;
	use Data::Dumper;
	use Scalar::Util qw(refaddr);

	my %topics;

	helper add_topic_subscriber => sub {
	my ($self, $topic) = @_;
	$topics{$topic}{refaddr $self} = $self;
	return;
	};

	helper remove_topic_subscriber => sub {
	my ($self, $topic) = @_;
	delete $topics{$topic}{refaddr $self};
	return;
	};

	helper publish_topic_message => sub {
	my ($self, $topic, $msg) = @_;
	$_->send($msg) for values %{ $topics{$topic} };
	return;
	};

	# Template with browser-side code
	get '/' => 'index';

	# WebSocket echo service
	websocket '/echo' => sub {
	my $c = shift;
	# Opened
	$c->app->log->debug('WebSocket opened');
	# Increase inactivity timeout for connection a bit
	$c->inactivity_timeout(120);
	# Incoming message
	$c->on(message => sub {
	my ($c, $msg) = @_;
	$c->send("echo: $msg");
	});
	# Closed
	$c->on(finish => sub {
	my ($c, $code, $reason) = @_;
	$c->app->log->debug("WebSocket closed with status $code");
	});
	};

	websocket '/pub/*topic' => sub {
	my $c = shift;
	$c->inactivity_timeout(120);
	my $topic = $c->stash('topic');
	$c->app->log->debug($c->tx->remote_address.' publish to '.$topic);
	$c->on(message => sub {
	my ($c, $msg) = @_;
	$c->publish_topic_message($topic, $msg);
	} );
	};

	websocket '/sub/*topic' => sub {
	my $c = shift;
	$c->inactivity_timeout(120);
	my $topic = $c->stash('topic');
	$c->add_topic_subscriber($topic);
	$c->app->log->debug($c->tx->remote_address.' subscribed to '.$topic);
	$c->on(finish => sub {
	my $c = shift;
	$c->remove_topic_subscriber($topic);
	} );
	};

	app->secrets([join('', map(("a".."z","A".."Z","0".."9")[rand 62], 0..20))]);
	app->start;
	__DATA__

	@@ index.html.ep
	<!DOCTYPE html>
	<html>
	<head><title>Echo</title></head>
	<body>
	<script>
	var count = 1;
	var pub = new WebSocket('<%= $self->url_for('pub/blah')->to_abs->scheme('ws') %>');
	var sub = new WebSocket('<%= $self->url_for('sub/blah')->to_abs->scheme('ws') %>');
	sub.onmessage = function (event) {
	document.body.appendChild(document.createTextNode(event.data+' '));
	};
	pub.onopen = function (event) {
	window.setInterval(function () { pub.send(count++) }, 1000);
	};
	</script>
	</body>
	</html>