view release on metacpan or  search on metacpan

lib/Alice/HTTP/Stream/XHR.pm  view on Meta::CPAN

has min_bytes => (
  is => 'ro',
  default => 1024,
);

sub BUILD {
  my $self = shift;

  my $local_time = time;
  my $remote_time = $self->start_time || $local_time;
  $self->offset($local_time - $remote_time);

  # better way to get the AE handle?
  my $hdl = $self->writer->{handle};

  $hdl->{rbuf_max} = 1024 * 10;

  my $close = sub {
    $self->close;
    undef $hdl;
    $self->on_error->();
  };

  $hdl->on_eof($close);
  $hdl->on_error($close);

  $self->send([{type => "identify", id => $self->id}]);
}

sub send {
  my ($self, $messages) = @_;
  return if $self->closed;

  $messages = [$messages] if $messages and ref $messages ne "ARRAY";

  $self->enqueue(@$messages) if $messages and @$messages;
  return if $self->delayed or $self->queue_empty;

  if (my $delay = $self->flooded) {
    $self->delay($delay);
    return;
  }
  $self->send_raw( $self->to_string );
  $self->last_send(time);
  $self->flush;
}

sub send_raw {
  my ($self, $string) = @_;

  my $output;

  if (! $self->started) {
    $output .= "--$separator\n";
    $self->started(1);
  }
  
  $output .= $string;

  $output .= "\n--$separator\n"
          .  " " x ($self->min_bytes - length $output);


  $self->writer->write( $output );
}

sub ping {
  my $self = shift;
  return if $self->closed;
  $self->send([{type => "action", event => "ping"}]);
}

sub close {
  my $self = shift;
  $self->flush;
  $self->writer->close if $self->writer;
  $self->writer(undef);
  $self->timer(undef);
  $self->closed(1);
}

sub flooded {
  my $self = shift;
  my $diff = time - $self->last_send;
  if ($diff < 0.2) {
    return 0.2 - $diff;
  }
  return 0;
}

sub delay {
  my ($self, $delay) = @_;
  $self->delayed(1);
  $self->timer(AnyEvent->timer(
    after => $delay,
    cb    => sub {
      $self->delayed(0);
      $self->timer(undef);
      $self->send;
    },
  ));
}

sub flush {
  my $self = shift;
  $self->clear_queue;
  $self->last_send(time);
}

sub to_string {
  my $self = shift;

  return to_json({
    queue => $self->queue,
    time  => time - $self->offset,
  }, {utf8 => 1, shrink => 1});
}

__PACKAGE__->meta->make_immutable;
1;