Add limit combinator to Stream

[sam-tombury]

What

Adds a limit combinator to Stream, which emits the first n elements and raises an error if there are more elements.

Why

I've implemented this a few times recently, and it seems to be sufficiently general to be worth implementing efficiently as a canoncial combinator. Example use case might be a middleware which enforces size limits on streams.

I'm not sure what the desired approach is to combinator generality - so it's completely fine if the consensus here is that it's not useful enough to warrant being included in the Stream API. Also hppy to rework the naming if desired, alternatives might be limitOrError or something like maxSize.

---

Thanks for opening a PR!

Target the main branch if this PR is targeted for the 3.x series and the series/2.5.x branch if targeted for the 2.5.x series.

If the CI build fails due to Scalafmt, run sbt scalafmtAll and push the changes (generally a good idea to run this prior to opening the PR).

Feel free to ask questions on the fs2 and fs2-dev channels on the Typelevel Discord server.

[diesalbla]

@sgjbryan Good evening, Sam. Thanks for the PR. As it happens, something like this is also included in the http4s library, and it certainly seems a desirable feature, so this should be a nice addition.

The name limit is a bit ambiguous, though. How would you feel about restrainOutputSizeTo(maxSize: Long)?

Finally, rather than issuing an IllegalState exception, could we perhaps issue a more specific one, such as StreamExceededSize or TooLargeStream, or similar?

[sam-tombury]

Great, thanks for the quick review!

It seems to me that the imperative tone of restrainOutputSizeTo isn't really in keeping with the naming conventions of the other combinators - how about limitLength or limitSize? I'd be quite content with any of {limit|restrict|constrain}{Length|Size}.

I would normally agree with raising a more specific exception, but it didn't appear to be the convention here (and I couldn't find any natural place to define a narrower exception). In principle I'd be happy to introduce something like fs2.error.StreamExceedSizeException extends IllegalStateException if that would be the preferred approach

[sam-tombury]

Just done some digging and I'm guessing this is the equivalent http4s implementation - thanks for mentioning that

[mpilquist]

I don't think we should add a custom exception type for this. We don't do it for other combinators with similar error cases (e.g. lastOrError).

[diesalbla]

It seems to me that the imperative tone of restrainOutputSizeTo isn't really in keeping with the naming conventions of the other combinators - how about limitLength or limitSize? I'd be quite content with any of {limit|restrict|constrain}{Length|Size}.

Well, take, filter, collect, or map, or zipWith are also themselves verbs, even though they all have the semantics not of modifying the immutable description (the immutable Stream object), but of creating a new immutable objects that wraps around the initial one. So it is not really a break from the rest. :)

The other reason for that is that this new method is a bit like an assertion, like Scala require statement, in that you weave into the new Stream the precondition "this source stream must be at most N elements long", hence the use of the verb restrain.

[sam-tombury]

That's true, I was probably a bit sloppy calling it imperative - I think it's probably the fluency (for want of a better word) that seems a bit out of place to me. That being said, there are examples of both (e.g. combinators like dropLastIf and delayBy are fluent when read with their arguments, but ones like interleave or concurrently aren't interleaveWith or concurrentlyWith).

To me, limit feels like more natural language than restrain - I think users would be able to find limitSize via IDE autocomplete but I personally don't think restrainOutputSize is particularly discoverable (other than if a user searches for size). In terms of pure discoverability I think limitOrError or takeOrError is probably most consistent.

I don't have a strong opinion though, so if general opinion is with restrainOutputSizeTo then happy to go with that

[mpilquist]

limit seems fine to me given operations like chunkLimit.