A lot of operators don't check they actually have a Scannable for scanUnsafe(Attr<Scannable>)

[simonbasle]

Expected behavior

All scanUnsafe implementations should return a Scannable when called with PARENT or ACTUAL.

Actual behavior

A lot of operators don't necessarily check that this is the case. They will happily return e.g. a CoreSubscriber, which happens to have mostly Scannable implementations (including the ones used in tests). But if you create an operator with an actual bare CoreSubsriber, scanning it for ACTUAL will fail with a ClassCastException.

Steps to reproduce

@Test
public void scanFail() {
    Scannable scannable = new FluxReplay<>(subscriber -> { }, 2, 2, null);
    Scannable parent = scannable.scan(Scannable.Attr.PARENT); // <-- ClassCastException

    assertThat(parent).isNotNull();
    assertThat(parent.isScanAvailable()).isFalse();
}

Reactor Core version

3.1.0.RELEASE

[simonbasle]

Backward-compatible solution design
We could work around the issue without having to go through the whole lot of operators, and provide better future-proof safety, by using Scannable.from.

When encountering these Attr<Scannable>, Scannable#scan could perform a safe conversion of the element from scanUnsafe with Scannable#from.

This can be generalized in Attr by adding a tryConvert(Object) method to the attributes. By default it would just cast as before, but scan would delegate that behavior to the Attr. For attributes that have a better strategy (like applying Scannable#from and getting a Null Object), said strategy could be defined at the attribute level.

In order to still be able to get the raw value from operators for these cases, a new RawAttr extends Attr<Object> could be introduced. They would wrap a typed Attr original and Scannable#scan would detect these and directly return the result of scanUnsafe(original).