+++ /dev/null
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-library thrift.test.transport.t_framed_transport_test;
-
-import 'dart:async';
-import 'dart:typed_data' show Uint8List;
-
-import 'package:dart2_constant/convert.dart' show utf8;
-import 'package:test/test.dart';
-import 'package:thrift/thrift.dart';
-
-void main() {
- group('TFramedTransport partial reads', () {
- final flushAwaitDuration = new Duration(seconds: 10);
-
- FakeReadOnlySocket socket;
- TSocketTransport socketTransport;
- TFramedTransport transport;
- var messageAvailable;
-
- setUp(() {
- socket = new FakeReadOnlySocket();
- socketTransport = new TClientSocketTransport(socket);
- transport = new TFramedTransport(socketTransport);
- messageAvailable = false;
- });
-
- expectNoReadableBytes() {
- var readBuffer = new Uint8List(128);
- var readBytes = transport.read(readBuffer, 0, readBuffer.lengthInBytes);
- expect(readBytes, 0);
- expect(messageAvailable, false);
- }
-
- test('Test transport reads messages where header and body are sent separately', () async {
- // buffer into which we'll read
- var readBuffer = new Uint8List(10);
- var readBytes;
-
- // registers for readable bytes
- var flushFuture = transport.flush().timeout(flushAwaitDuration);
- flushFuture.then((_) {
- messageAvailable = true;
- });
-
- // write header bytes
- socket.messageController.add(new Uint8List.fromList([0x00, 0x00, 0x00, 0x06]));
-
- // you shouldn't be able to get any bytes from the read,
- // because the header has been consumed internally
- expectNoReadableBytes();
-
- // write first batch of body
- socket.messageController.add(new Uint8List.fromList(utf8.encode("He")));
-
- // you shouldn't be able to get any bytes from the read,
- // because the frame has been consumed internally
- expectNoReadableBytes();
-
- // write second batch of body
- socket.messageController.add(new Uint8List.fromList(utf8.encode("llo!")));
-
- // have to wait for the flush to complete,
- // because it's only then that the frame is available for reading
- await flushFuture;
- expect(messageAvailable, true);
-
- // at this point the frame is complete, so we expect the read to complete
- readBytes = transport.read(readBuffer, 0, readBuffer.lengthInBytes);
- expect(readBytes, 6);
- expect(readBuffer.sublist(0, 6), utf8.encode("Hello!"));
- });
-
- test('Test transport reads messages where header is sent in pieces '
- 'and body is also sent in pieces', () async {
- // buffer into which we'll read
- var readBuffer = new Uint8List(10);
- var readBytes;
-
- // registers for readable bytes
- var flushFuture = transport.flush().timeout(flushAwaitDuration);
- flushFuture.then((_) {
- messageAvailable = true;
- });
-
- // write first part of header bytes
- socket.messageController.add(new Uint8List.fromList([0x00, 0x00]));
-
- // you shouldn't be able to get any bytes from the read
- expectNoReadableBytes();
-
- // write second part of header bytes
- socket.messageController.add(new Uint8List.fromList([0x00, 0x03]));
-
- // you shouldn't be able to get any bytes from the read again
- // because only the header was read, and there's no frame body
- readBytes = expectNoReadableBytes();
-
- // write first batch of body
- socket.messageController.add(new Uint8List.fromList(utf8.encode("H")));
-
- // you shouldn't be able to get any bytes from the read,
- // because the frame has been consumed internally
- expectNoReadableBytes();
-
- // write second batch of body
- socket.messageController.add(new Uint8List.fromList(utf8.encode("i!")));
-
- // have to wait for the flush to complete,
- // because it's only then that the frame is available for reading
- await flushFuture;
- expect(messageAvailable, true);
-
- // at this point the frame is complete, so we expect the read to complete
- readBytes = transport.read(readBuffer, 0, readBuffer.lengthInBytes);
- expect(readBytes, 3);
- expect(readBuffer.sublist(0, 3), utf8.encode("Hi!"));
- });
- });
-}
-
-
-
-class FakeReadOnlySocket extends TSocket {
-
- StreamController<Uint8List> messageController = new StreamController<Uint8List>(sync: true);
- StreamController<Object> errorController = new StreamController<Object>();
- StreamController<TSocketState> stateController = new StreamController<TSocketState>();
-
- @override
- Future close() {
- // noop
- }
-
- @override
- bool get isClosed => false;
-
- @override
- bool get isOpen => true;
-
- @override
- Stream<Object> get onError => errorController.stream;
-
- @override
- Stream<Uint8List> get onMessage => messageController.stream;
-
- @override
- Stream<TSocketState> get onState => stateController.stream;
-
- @override
- Future open() {
- // noop
- }
-
- @override
- void send(Uint8List data) {
- // noop
- }
-}
-
projects / ceph.git / blobdiff