view release on metacpan or  search on metacpan

t/anyevent_websocket_connection.t  view on Meta::CPAN

  my $y_recv = $cv_b_recv->recv;
  my $parser = Protocol::WebSocket::Frame->new;
  $parser->append($y_recv);
  ok defined($parser->next_bytes), "received a complete frame";
  ok $parser->is_close, "... and it's a close frame";

};

subtest 'Connection should refuse extremely huge messages' => sub {

  subtest "Connection should refuse huge frames", sub {

    my ($x_conn, $y_handle) = create_connection_and_handle();
    my $cv_finish = AnyEvent->condvar;
    $cv_finish->begin;
    $cv_finish->begin;
    my @received_messages = ();
    $x_conn->on(finish => sub {
      $cv_finish->end;
    });
    $x_conn->on(each_message => sub {
      push(@received_messages, $_[1]);
    });
    $y_handle->on_error(sub {
      my $handle = shift;
      $handle->push_shutdown;
      $cv_finish->end;
    });
    $y_handle->on_read(sub { });

    my $frame_header = pack("H*", "827f00000000ffffffff"); # frame payload size = 2**32 - 1 bytes
    my $MAX_SEND_PAYLOAD = 1024; # for safety
    my $count_send_payload = 0;
    $y_handle->push_write($frame_header);
    $y_handle->on_drain(sub {
      my $handle = shift;
      $count_send_payload++;
      if($count_send_payload >= $MAX_SEND_PAYLOAD)
      {
        fail("Connection should be aborted by now.");
        $handle->on_drain(undef);
        $handle->push_shutdown;
        $cv_finish->send;
        return;
      }

      # push_write is delayed to prevent deep-recursion and to give
      # $x_conn chance to receive data.
      my $w; $w = AnyEvent->idle(cb => sub {
        undef $w;
        $handle->push_write("A" x 256);
      });
    });
    $cv_finish->recv;

    is scalar(@received_messages), 0, "the frame is too huge to receive.";
  };


  subtest "Connection should refuse messages with too many fragments", sub {
    my ($x_conn, $y_handle) = create_connection_and_handle;
    my $cv_finish = AnyEvent->condvar;
    $cv_finish->begin;
    $cv_finish->begin;
    my @received_messages = ();
    $x_conn->on(finish => sub {
      $cv_finish->end;
    });
    $x_conn->on(each_message => sub {
      push(@received_messages, $_[1])
    });
    $y_handle->on_error(sub {
      my $handle = shift;
      $handle->push_shutdown;
      $cv_finish->end;
    });
    $y_handle->on_read(sub {});

    my $MAX_SEND_FRAMES = 10000;
    my $count_send_frame = 0;
    $y_handle->push_write(Protocol::WebSocket::Frame->new(fin => 0, opcode => 1, buffer => "A")->to_bytes);
    $y_handle->on_drain(sub {
      my $handle = shift;
      $count_send_frame++;
      if($count_send_frame >= $MAX_SEND_FRAMES)
      {
        fail("Connection should be aborted by now.");
        $handle->on_drain(undef);
        $handle->push_shutdown;
        $cv_finish->send;
        return;
      }
      my $w; $w = AnyEvent->idle(cb => sub {
        undef $w;
        $handle->push_write(Protocol::WebSocket::Frame->new(fin => 0, opcode => 0, buffer => "A")->to_bytes);
      });
    });
    $cv_finish->recv;

    is scalar(@received_messages), 0, "the message consists of too many fragments to receive.";
  };
};

subtest 'other end is closed' => sub {

  my($x,$y) = create_connection_pair;

  my $round_trip = sub {

    my($message) = @_;

    my $done = AnyEvent->condvar;

    $y->on(next_message => sub {
      my(undef, $message) = @_;
      $done->send($message);
    });

    $x->send($message);

    $done->recv;

  };

  my $closed = 0;

  my $quit_cv = AnyEvent->condvar;
  $y->on(finish => sub {
    $closed = 1;
    $quit_cv->send("finished");
  });

  is(
    $round_trip->('a'),
    object {
      call decoded_body => 'a';
    },
    'single character',
  );

  is(
    $round_trip->('quit'),
    object {
      call decoded_body => 'quit';
    },
    'quit',
  );

  $x->close;

  $quit_cv->recv;

  is $closed, 1, "closed";

};

subtest 'close codes' => sub {

  my @test_data = (
    [ [],                 [1000, ''],         'empty list defaults to 1005'     ],