ceph/src/jaegertracing/thrift/lib/dart/lib/src/browser/t_web_socket.dart

   1 /// Licensed to the Apache Software Foundation (ASF) under one
   2 /// or more contributor license agreements. See the NOTICE file
   3 /// distributed with this work for additional information
   4 /// regarding copyright ownership. The ASF licenses this file
   5 /// to you under the Apache License, Version 2.0 (the
   6 /// "License"); you may not use this file except in compliance
   7 /// with the License. You may obtain a copy of the License at
   8 ///
   9 /// http://www.apache.org/licenses/LICENSE-2.0
  10 ///
  11 /// Unless required by applicable law or agreed to in writing,
  12 /// software distributed under the License is distributed on an
  13 /// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  14 /// KIND, either express or implied. See the License for the
  15 /// specific language governing permissions and limitations
  16 /// under the License.
  17
  18 library thrift.src.browser;
  19
  20 import 'dart:async';
  21 import 'package:dart2_constant/convert.dart' show base64;
  22 import 'dart:html' show CloseEvent;
  23 import 'dart:html' show Event;
  24 import 'dart:html' show MessageEvent;
  25 import 'dart:html' show WebSocket;
  26 import 'dart:typed_data' show Uint8List;
  27
  28 import 'package:thrift/thrift.dart';
  29
  30 /// A [TSocket] backed by a [WebSocket] from dart:html
  31 class TWebSocket implements TSocket {
  32   final Uri url;
  33
  34   final StreamController<TSocketState> _onStateController;
  35   Stream<TSocketState> get onState => _onStateController.stream;
  36
  37   final StreamController<Object> _onErrorController;
  38   Stream<Object> get onError => _onErrorController.stream;
  39
  40   final StreamController<Uint8List> _onMessageController;
  41   Stream<Uint8List> get onMessage => _onMessageController.stream;
  42
  43   final List<Uint8List> _requests = [];
  44
  45   TWebSocket(this.url)
  46       : _onStateController = new StreamController.broadcast(),
  47         _onErrorController = new StreamController.broadcast(),
  48         _onMessageController = new StreamController.broadcast() {
  49     if (url == null || !url.hasAuthority || !url.hasPort) {
  50       throw new ArgumentError('Invalid url');
  51     }
  52   }
  53
  54   WebSocket _socket;
  55
  56   bool get isOpen => _socket != null && _socket.readyState == WebSocket.OPEN;
  57
  58   bool get isClosed =>
  59       _socket == null || _socket.readyState == WebSocket.CLOSED;
  60
  61   Future open() {
  62     if (!isClosed) {
  63       throw new TTransportError(
  64           TTransportErrorType.ALREADY_OPEN, 'Socket already connected');
  65     }
  66
  67     _socket = new WebSocket(url.toString());
  68     _socket.onError.listen(_onError);
  69     _socket.onOpen.listen(_onOpen);
  70     _socket.onClose.listen(_onClose);
  71     _socket.onMessage.listen(_onMessage);
  72
  73     return _socket.onOpen.first;
  74   }
  75
  76   Future close() {
  77     if (_socket != null) {
  78       _socket.close();
  79       return _socket.onClose.first;
  80     } else {
  81       return new Future.value();
  82     }
  83   }
  84
  85   void send(Uint8List data) {
  86     _requests.add(data);
  87     _sendRequests();
  88   }
  89
  90   void _sendRequests() {
  91     while (isOpen && _requests.isNotEmpty) {
  92       Uint8List data = _requests.removeAt(0);
  93       _socket.sendString(base64.encode(data));
  94     }
  95   }
  96
  97   void _onOpen(Event event) {
  98     _onStateController.add(TSocketState.OPEN);
  99     _sendRequests();
 100   }
 101
 102   void _onClose(CloseEvent event) {
 103     _socket = null;
 104
 105     if (_requests.isNotEmpty) {
 106       _onErrorController
 107           .add(new StateError('Socket was closed with pending requests'));
 108     }
 109     _requests.clear();
 110
 111     _onStateController.add(TSocketState.CLOSED);
 112   }
 113
 114   void _onMessage(MessageEvent message) {
 115     try {
 116       Uint8List data = new Uint8List.fromList(base64.decode(message.data));
 117       _onMessageController.add(data);
 118     } on FormatException catch (_) {
 119       var error = new TProtocolError(TProtocolErrorType.INVALID_DATA,
 120           "Expected a Base 64 encoded string.");
 121       _onErrorController.add(error);
 122     }
 123   }
 124
 125   void _onError(Event event) {
 126     close();
 127     _onErrorController.add(event.toString());
 128   }
 129 }