pkgs/http2/test/src/streams/simple_flow_test.dart

// Copyright (c) 2015, the Dart project authors.  Please see the AUTHORS file
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.

import 'dart:async';

import 'package:http2/transport.dart';
import 'package:test/test.dart';

import 'helper.dart';

void main() {
  group('streams', () {
    group('flowcontrol', () {
      const numOfOneKB = 1000;

      var expectedHeaders = [Header.ascii('key', 'value')];
      var allBytes = List.generate(numOfOneKB * 1024, (i) => i % 256);
      allBytes.addAll(List.generate(42, (i) => 42));

      void Function(StreamMessage) headersTestFun(String type) {
        return expectAsync1((StreamMessage msg) {
          expect(
              msg,
              isA<HeadersStreamMessage>()
                  .having((m) => m.headers.first.name, 'headers.first.name',
                      expectedHeaders.first.name)
                  .having((m) => m.headers.first.value, 'headers.first.value',
                      expectedHeaders.first.value));
        });
      }

      var serverReceivedAllBytes = Completer<void>();

      void Function(StreamMessage) messageTestFun(String type) {
        var expectHeader = true;
        var numBytesReceived = 0;
        return (StreamMessage msg) {
          if (expectHeader) {
            expectHeader = false;
            expect(
                msg,
                isA<HeadersStreamMessage>()
                    .having((m) => m.headers.first.name, 'headers.first.name',
                        expectedHeaders.first.name)
                    .having((m) => m.headers.first.value, 'headers.first.value',
                        expectedHeaders.first.value));
          } else {
            expect(msg, isA<DataStreamMessage>());
            var bytes = (msg as DataStreamMessage).bytes;
            expect(
                bytes,
                allBytes.sublist(
                    numBytesReceived, numBytesReceived + bytes.length));
            numBytesReceived += bytes.length;

            if (numBytesReceived > allBytes.length) {
              if (serverReceivedAllBytes.isCompleted) {
                throw Exception('Got more messages than expected');
              }
              serverReceivedAllBytes.complete();
            }
          }
        };
      }

      void sendData(TransportStream cStream) {
        for (var i = 0; i < (allBytes.length + 1023) ~/ 1024; i++) {
          var end = 1024 * (i + 1);
          var isLast = end > allBytes.length;
          if (isLast) {
            end = allBytes.length;
          }
          cStream.sendData(allBytes.sublist(1024 * i, end), endStream: isLast);
        }
      }

      streamTest('single-header-request--empty-response',
          (ClientTransportConnection client,
              ServerTransportConnection server) async {
        server.incomingStreams
            .listen(expectAsync1((TransportStream sStream) async {
          sStream.incomingMessages
              .listen(messageTestFun('server'), onDone: expectAsync0(() {}));
          sStream.sendHeaders(expectedHeaders, endStream: true);
          await serverReceivedAllBytes.future;
        }));

        TransportStream cStream = client.makeRequest(expectedHeaders);
        sendData(cStream);
        cStream.incomingMessages
            .listen(headersTestFun('client'), onDone: expectAsync0(() {}));
      });
    });
  });
}