lib/Tak/JSONChannel.pm

   1 package Tak::JSONChannel;
   2
   3 use JSON::PP qw(encode_json decode_json);
   4 use IO::Handle;
   5 use Scalar::Util qw(weaken);
   6 use Moo;
   7
   8 has read_fh => (is => 'ro', required => 1);
   9 has write_fh => (is => 'ro', required => 1);
  10
  11 sub BUILD { shift->write_fh->autoflush(1); }
  12
  13 sub receive {
  14   my ($self) = @_;
  15   while (my $line = readline($self->read_fh)) {
  16     if (my $unpacked = $self->_unpack_line($line)) {
  17       return $unpacked;
  18     }
  19   }
  20 }
  21
  22 sub _unpack_line {
  23   my ($self, $line) = @_;
  24   my $data = eval { decode_json($line) };
  25   unless ($data) {
  26     $self->send(MISTAKE => invalid_json => $@||'No data and no exception');
  27     return;
  28   }
  29   unless (ref($data) eq 'ARRAY') {
  30     $self->send(MISTAKE => message_format => "Not an ARRAY");
  31     return;
  32   }
  33   unless (@$data > 0) {
  34     $self->send(MISTAKE => message_format => "Empty request array");
  35     return;
  36   }
  37   $data;
  38 }
  39
  40 sub send {
  41   my ($self, @msg) = @_;
  42   my $json = eval { encode_json(\@msg) };
  43   unless ($json) {
  44     $self->_raw_send(
  45       encode_json(
  46         [ FAILURE => invalid_message => $@||'No data and no exception' ]
  47       )
  48     );
  49     return;
  50   }
  51   $self->_raw_send($json);
  52 }
  53
  54 sub _raw_send {
  55   my ($self, $raw) = @_;
  56 #warn "Sending: ${raw}\n";
  57   print { $self->write_fh } $raw."\n";
  58 }
  59
  60 1;