[Streaming Generator] Make it compatible with wait

[rkooo567]

Why are these changes needed?

This PR makes the streaming generator compatible with ray.wait.

The semantic is as follows;

def f():
    for _ in range(3):
        yield 1
generator = f.options(num_returns="streaming").remote()
# The generator will be in ready if the next reference is available. Otherwise it is in unready.
# This should work with all other options from ray.wait (including fetch_local=True/False)
ready, unready = ray.wait([generator])

# if the generator's next ref is not ready in 0.1 second, it will be in unready.
# otherwise, it is in ready
ready, unready = ray.wait([generator], timeout=0.1)

# If the generator's next ref is available, it is considered as 1 return
# In this case, this will return if both generator and ref is ready.
ready, unready = ray.wait([generator, ref], num_returns=2)

# if the generator's next ref is available, it will fetch the object to the local node
ready, unready = ray.wait([generator, ref], fetch_local=True)

From the previous PR #36070, we are now able to peek the object reference, and the peeked object is guaranteed to be resolved. We can always peek the next object from the generator and wait on that reference to make the generator compatible to ray.wait.

Related issue number

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(