Fix multipart transformer in package:mime

dart/pkg/mime/lib/src/bound_multipart_stream.dart

 // Copyright (c) 2014, 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.
 library mime.bound_multipart_stream;
 
 import 'dart:async';
 import 'dart:convert';
 
 import 'mime_shared.dart';
 import 'char_code.dart';
@@ skipping 56 matching lines @@
    static const int _LAST_BOUNDARY_DASH2 = 11;
    static const int _LAST_BOUNDARY_ENDING = 12;
    static const int _LAST_BOUNDARY_END = 13;
    static const int _DONE = 14;
    static const int _FAIL = 15;
 
    final List<int> _boundary;
    final List<int> _headerField = [];
    final List<int> _headerValue = [];
 
+   // The following states belong to `_controller`, state changes will not be
+   // immediately acted upon but rather only after the current
+   // `_multipartController` is done.
+   static const int _CONTROLLER_STATE_IDLE = 0;
+   static const int _CONTROLLER_STATE_ACTIVE = 1;
+   static const int _CONTROLLER_STATE_PAUSED = 2;
+   static const int _CONTROLLER_STATE_CANCELLED = 3;

[Lasse Reichstein Nielsen, 2015/02/12 13:40:04]

The US version is CANCELED with one L (as I have recently learned).

[kustermann, 2015/02/12 13:51:24]

Good to know :)

Done.

+
+   int _controllerState = _CONTROLLER_STATE_IDLE;
+
    StreamController _controller;
 
    Stream<MimeMultipart> get stream => _controller.stream;
 
    StreamSubscription _subscription;
 
    StreamController _multipartController;
    Map<String, String> _headers;
 
    int _state = _START;
    int _boundaryIndex = 2;
 
    // Current index in the data buffer. If index is negative then it
    // is the index into the artificial prefix of the boundary string.
    int _index;
    List<int> _buffer;
 
    BoundMultipartStream(this._boundary, Stream<List<int>> stream) {
      _controller = new StreamController(
          sync: true,
          onPause: _pauseStream,
          onResume:_resumeStream,
          onCancel: () {
-           _subscription.cancel();
+           _controllerState = _CONTROLLER_STATE_CANCELLED;
+           _tryPropagateControllerState();
          },
          onListen: () {
+           _controllerState = _CONTROLLER_STATE_ACTIVE;
            _subscription = stream.listen(
                (data) {
                  assert(_buffer == null);
-                 _pauseStream();
+                 _subscription.pause();
                  _buffer = data;
                  _index = 0;
                  _parse();
                },
                onDone: () {
                  if (_state != _DONE) {
                    _controller.addError(
                        new MimeMultipartException("Bad multipart ending"));
                  }
                  _controller.close();
                },
                onError: _controller.addError);
          });
    }
 
    void _resumeStream() {
-     _subscription.resume();
+     assert (_controllerState == _CONTROLLER_STATE_PAUSED);
+     _controllerState = _CONTROLLER_STATE_ACTIVE;
+     _tryPropagateControllerState();
    }
 
    void _pauseStream() {
-     _subscription.pause();
+     _controllerState = _CONTROLLER_STATE_PAUSED;
+     _tryPropagateControllerState();
    }
 
+   void _tryPropagateControllerState() {
+     if (_multipartController == null) {
+       switch (_controllerState) {
+         case _CONTROLLER_STATE_ACTIVE:
+           if (_subscription.isPaused) _subscription.resume();
+           break;
+         case _CONTROLLER_STATE_PAUSED:
+           if (!_subscription.isPaused) _subscription.pause();
+           break;
+         case _CONTROLLER_STATE_CANCELLED:
+           _subscription.cancel();
+            break;
+         default:
+           throw new Exception("This code should never be reached.");

[Lasse Reichstein Nielsen, 2015/02/12 13:40:04]

This should be an Error, not an Exception.
I suggest a StateError.

[kustermann, 2015/02/12 13:51:24]

Done.

I was looking for something like `throw new UnreachableError()` but I could only
find `throw new UnimplementedError()`

+       }
+     }
+   }
 
    void _parse() {
      // Number of boundary bytes to artificially place before the supplied data.
      int boundaryPrefix = 0;
      // Position where content starts. Will be null if no known content
      // start exists. Will be negative of the content starts in the
      // boundary prefix. Will be zero or position if the content starts
      // in the current buffer.
      int contentStartIndex;
 
@@ skipping 63 matching lines @@
            } else {
              _expectWhitespace(byte);
            }
            break;
 
          case _BOUNDARY_END:
            _expectByteValue(byte, CharCode.LF);
            if (_multipartController != null) {
              _multipartController.close();
              _multipartController = null;
+             _tryPropagateControllerState();
            }
            _state = _HEADER_START;
            break;
 
          case _HEADER_START:
            _headers = new Map<String, String>();
            if (byte == CharCode.CR) {
              _state = _HEADER_ENDING;
            } else {
              // Start of new header field.
@@ skipping 52 matching lines @@
                _headerField.add(_toLowerCase(byte));
                _state = _HEADER_FIELD;
              }
            }
            break;
 
          case _HEADER_ENDING:
            _expectByteValue(byte, CharCode.LF);
            _multipartController = new StreamController(
                sync: true,
-               onPause: () {
-                 _pauseStream();
-               },
+               onPause: _subscription.pause,
                onResume: () {
-                 _resumeStream();
+                 _subscription.resume();
                  _parse();
                });
            _controller.add(
                new _MimeMultipart(_headers, _multipartController.stream));
            _headers = null;
            _state = _CONTENT;
            contentStartIndex = _index + 1;
            break;
 
          case _CONTENT:
            if (byte == _boundary[_boundaryIndex]) {
              _boundaryIndex++;
              if (_boundaryIndex == _boundary.length) {
                if (contentStartIndex != null) {
                  _index++;
                  reportData();
                  _index--;
                }
                _multipartController.close();
+               _multipartController = null;
+               _tryPropagateControllerState();
                _boundaryIndex = 0;
                _state = _BOUNDARY_ENDING;
              }
            } else {
              // Restart matching of the boundary.
              _index = _index - _boundaryIndex;
              if (contentStartIndex == null) contentStartIndex = _index;
              _boundaryIndex = 0;
            }
            break;
 
          case _LAST_BOUNDARY_DASH2:
            _expectByteValue(byte, CharCode.DASH);
            _state = _LAST_BOUNDARY_ENDING;
            break;
 
          case _LAST_BOUNDARY_ENDING:
            if (byte == CharCode.CR) {
              _state = _LAST_BOUNDARY_END;
            } else {
              _expectWhitespace(byte);
            }
            break;
 
          case _LAST_BOUNDARY_END:
            _expectByteValue(byte, CharCode.LF);
            if (_multipartController != null) {
              _multipartController.close();
              _multipartController = null;
+             _tryPropagateControllerState();
            }
            _state = _DONE;
            break;
 
          default:
            // Should be unreachable.
            assert(false);
            break;
        }
 
        // Move to the next byte.
        _index++;
      }
 
      // Report any known content.
      if (_state == _CONTENT && contentStartIndex != null) {
        reportData();
      }
 
      // Resume if at end.
      if (_index == _buffer.length) {
        _buffer = null;
        _index = null;
-       _resumeStream();
+       _subscription.resume();
      }
    }
 }