Better "collect Strict" specifically for WebSockets Flow

**Issue by [ktoso](https://github.com/ktoso)**
_Monday Mar 21, 2016 at 09:59 GMT_
_Originally opened as https://github.com/akka/akka/issues/20096_

---

This is what people do (and I'd do so myself):

``` scala
val flow = Flow[Message].collect {
case TextMessage.Strict(msg) =>
msg.parseJson.convertTo[Anything]
```

which can randomly break "almost work", due to even the smallest message sometimes ending up as `Streamed` (I've seen this with msg as small as `"Hello world"`).

"Almost work" here is defined as – small message comes in, but ends up being streamed, the collect here drops it, processing is stalled until subscription timeout triggers killing the `textStream` of the streamed message.

We should provide a safe way to "I know these will be strict, or force them to be".
This is somewhat related to the "auto drain" feature for HttpRequests too, however likely a different trick will have to be applied.

For reference, this is what I ended up with in a project I hackked on:

``` scala
implicit class flowTweaks[M](val wsInput: Source[Message, M]) {
def forceTextStrict: Source[Strict, M] = wsInput
.collect {
case TextMessage.Strict(text) ⇒ Future.successful(text)
case TextMessage.Streamed(textStream) ⇒ textStream.runFold("")(_ + _)
case BinaryMessage.Strict(binary) ⇒ Skip
case BinaryMessage.Streamed(binaryStream) ⇒ binaryStream.runWith(Sink.ignore); Skip
}
.filterNot(_ == Skip)
.mapAsync(1)(ConstantFun.scalaIdentityFunction)
.map(TextMessage.Strict)
}
```

Start by reading the WebSocket Flow example using `Flow[Message].collect` and the issue's `TextMessage.Strict`/`TextMessage.Streamed` example to understand where streamed messages get dropped. The issue proposes a safe way to handle or force strict messages but does not name files or tests; done means providing that safe behavior for WebSocket flows.

Written by the indexing model from the issue text.