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 Log::Contextual qw(:log);
   7 use Moo;
   8
   9 has read_fh => (is => 'ro', required => 1);
  10 has write_fh => (is => 'ro', required => 1);
  11
  12 sub BUILD { shift->write_fh->autoflush(1); }
  13
  14 sub read_message {
  15   my ($self) = @_;
  16   if (defined(my $line = readline($self->read_fh))) {
  17     log_trace { "Received $line" };
  18     if (my $unpacked = $self->_unpack_line($line)) {
  19       return $unpacked;
  20     }
  21   } else {
  22     return [ 'close', 'channel' ];
  23   }
  24 }
  25
  26 sub _unpack_line {
  27   my ($self, $line) = @_;
  28   my $data = eval { decode_json($line) };
  29   unless ($data) {
  30     $self->write_message(mistake => invalid_json => $@||'No data and no exception');
  31     return;
  32   }
  33   unless (ref($data) eq 'ARRAY') {
  34     $self->write_message(mistake => message_format => "Not an ARRAY");
  35     return;
  36   }
  37   unless (@$data > 0) {
  38     $self->write_message(mistake => message_format => "Empty request array");
  39     return;
  40   }
  41   $data;
  42 }
  43
  44 sub write_message {
  45   my ($self, @msg) = @_;
  46   my $json = eval { encode_json(\@msg) };
  47   unless ($json) {
  48     $self->_raw_write_message(
  49       encode_json(
  50         [ failure => invalid_message => $@||'No data and no exception' ]
  51       )
  52     );
  53     return;
  54   }
  55   log_trace { "Sending: $json" };
  56   $self->_raw_write_message($json);
  57 }
  58
  59 sub _raw_write_message {
  60   my ($self, $raw) = @_;
  61 #warn "Sending: ${raw}\n";
  62   print { $self->write_fh } $raw."\n";
  63 }
  64
  65 1;