Optimize the **Java** Flow interop

[BalmungSan]

Follow up of #3102

• Reduce allocations on StreamSubscription
• Reduce allocations in the internal buffer of StreamSubscriber
• Cross compile (will be done in a separate PR WIP: Cross-compile interop.flow #3160)

[BalmungSan]

Some questions:

1. Was just moving the files enough for cross-compiling?
2. For some reason Scala 3 didn't like my covariant State so I had to make it invariant which was a real PITA.
3. I guess the ClassTag change is a non-go... but then, what should we do?

[armanbilge]

1. No, we should also run the tests to check that it actually works (although not the TCK ones, since those won't cross-compile). When I checked this the current API was using stuff that doesn't cross-compile. Not sure if that's still the case after your changes.

2. 🙃

3. Why do we need ClassTag now? The answer is, don't need it 😂

[armanbilge]

Is this canceled check really doing anything useful?

[armanbilge]

Oh, yes it is, this comment would almost be more helpful here.

fs2/core/jvm/src/main/scala/fs2/interop/flow/StreamSubscription.scala

Lines 124 to 129 in 1f6bc98

	// According to the spec, it's acceptable for a concurrent cancel to not
	// be processed immediately, but if you have synchronous
	// `cancel(); request()`,
	// then the request must be a NOOP.
	// See https://github.com/zainab-ali/fs2-reactive-streams/issues/29
	// and https://github.com/zainab-ali/fs2-reactive-streams/issues/46

[BalmungSan]

Well, that was the original place of the comment, I just polished it a little.
I can move it if you prefer, the thing is that it also affects cancel, and well both methods are close together.

[armanbilge]

The diff was horrible, but when viewed standalone the new implementations are quite elegant. Thanks for bearing with me 😇

[armanbilge]

I tossed and turned on this for several nights but in the end I think it's a win. Besides making it easy to use the Array-based buffer, it also allowed us to remove the type parameter from the state machine. Basically, in this unusual case I think threading the type parameter through the implementation was less ergonomic than "pre-erasing" it.