when:
        def received1 = receiver.deserialize(originator.serialize(original))
        def reply1 = originator.deserialize(receiver.serialize(received1))
        def received2 = receiver.deserialize(originator.serialize(original))
        def reply2 = originator.deserialize(receiver.serialize(received2))

        then:
        received1.class.classLoader == received2.class.classLoader

                    broadcaster.foo(it)
                }
            }
        }

        then:
        received1.size() == 60
        received2.size() == 60
        received1 == received2
        received1.findAll { it.startsWith("a") } == events1
        received1.findAll { it.startsWith("b") } == events2
    }

    List<String> events(String prefix, int count) {

import javax.annotation.Nullable;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * Receives and handles messages about a given worker action executed by a worker process.
 * <p>
 * This receiver is used per worker action.
 */
@NonNullApi
public class Receiver implements ResponseProtocol, StreamCompletion, StreamFailureHandler {
    private static final Object NULL = new Object();

 * A source for messages. Implementations do not have to be thread-safe.
 */
public interface Receive<T> {
    /**
     * Blocks until the next message is available. Returns null when the end of the message stream has been reached.
     *
     * @return The next message, or null when the end of the stream has been reached.
     */
    @Nullable
    T receive();

// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

// Test Implicit methods for embedded types and
// mixed pointer and non-pointer receivers.

package main

type T int

var nv, np int

func (t T) V() {
	if t != 42 {
		panic(t)
	}
	nv++
}

func (t *T) P() {
	if *t != 42 {
		println(t, *t)
		panic("fail")
	}

Robert Griesemer <******@****.***> 1606873032 -0800

// indicates whether the value is valid, or whether the Sender has
// been closed and no more values will be received.
func (r *Receiver[T]) Next() (T, bool) {
	v, ok := <-r.values
	return v, ok
}

// finalize is a finalizer for the receiver.
func (r *Receiver[T]) finalize() {
	close(r.done)

// indicates whether the value is valid, or whether the Sender has
// been closed and no more values will be received.
func (r *Receiver[T]) Next() (T, bool) {
	v, ok := <-r.values
	return v, ok
}

// finalize is a finalizer for the receiver.
func (r *Receiver[T]) finalize() {
	close(r.done)

func Ranger[T any]() (*Sender[T], *Receiver[T]) {
	c := make(chan T)
	d := make(chan bool)
	s := &Sender[T]{values: c, done: d}
	r := &Receiver[T]{values: c, done: d}
	runtime.SetFinalizer(r, r.finalize)
	return s, r
}

// A sender is used to send values to a Receiver.
type Sender[T any] struct {
	values chan<- T
	done   <-chan bool
}

	}
}

func TestRanger() {
	s, r := _Ranger[int]()

	ctx := context.Background()
	go func() {
		// Receive one value then exit.
		v, ok := r.Next(ctx)
		if !ok {
			panic(fmt.Sprintf("did not receive any values"))
		} else if v != 1 {
			panic(fmt.Sprintf("received %d, want 1", v))
		}
	}()

	c1 := make(chan bool)
	c2 := make(chan bool)
	go func() {
		defer close(c2)