"context"
	"fmt"
	"sync"
)

type status int

const (
	Pending status = iota
	Ready
	Stopped
)

// ready is a three state condition variable that blocks until is Ready if is not Stopped.
// Its initial state is Pending and its state machine diagram is as follow.
//
// Pending <------> Ready -----> Stopped
//
//	|                           ^
//	└---------------------------┘
type ready struct {

	f.Lock()
	defer f.Unlock()
	if !f.stopped {
		klog.V(4).Infof("Stopping fake watcher.")
		close(f.result)
		f.stopped = true
	}
}

func (f *FakeWatcher) IsStopped() bool {
	f.Lock()
	defer f.Unlock()
	return f.stopped
}

// Reset prepares the watcher to be reused.
func (f *FakeWatcher) Reset() {
	f.Lock()
	defer f.Unlock()
	f.stopped = false
	f.result = make(chan Event)
}

	m.incomingBlock.Lock()
	defer m.incomingBlock.Unlock()

	// Ensure that if the broadcaster is stopped we do not send events to it.
	select {
	case <-m.stopped:
		return false, fmt.Errorf("broadcaster already stopped")
	default:
	}

	select {
	case m.incoming <- Event{action, obj}:
		return true, nil
	default:
		return false, nil
	}
}

            executor.execute(action1)
            executor.execute(action2)
            executor.stop()
            instant.stopped
        }

        then:
        instant.stopped > instant.completed1
        instant.stopped > instant.completed2
    }

    def factoryStopBlocksUntilAllJobsAreComplete() {
        given:
        def action1 = {
            thread.block()

                            break;
                        case Stopped:
                            LOGGER.debug("daemon has stopped.");
                            return DaemonStopState.Clean;
                        case ForceStopped:
                            LOGGER.debug("daemon has been force stopped.");
                            return DaemonStopState.Forced;
                    }

        out =~ /\n\s*\d+\s+IDLE\s+([\w\.\+\-]+)/
        out =~ /\n\s*\d+\s+BUSY\s+([\w\.\+\-]+)/
        out =~ /\n\s*12346\s+STOPPED\s+\(GRACEFUL_EXPIRE_REASON\)/

        cleanup:
        block?.releaseAll()
        build?.waitForFinish()
    }

    def "reports stopped status of recently stopped daemons"() {
        given:

        expect:
        notStopped

        when: "stopped first time"
        coordinator.stop()

        then: "stops"
        stopped

        when: "requested again"
        coordinator.stop()

        then:
        stopped
        0 * _._
    }

    def "await idle timeout does nothing when already stopped"() {
        given:
        coordinator.stop()

        when:

                case STOPPING:
                    throw new IllegalStateException(String.format("Cannot use %s as it is currently stopping.", displayName));
                case STOPPED:
                    throw new IllegalStateException(String.format("Cannot use %s as it has been stopped.", displayName));
            }
            Integer depth = usages.get(Thread.currentThread());
            if (depth == null) {

            if (interrupted) {
                Thread.currentThread().interrupt();
            }
            if (state == State.Stopped) {
                throw new IllegalStateException("Cannot dispatch message, as this message dispatch has been stopped. Message: " + message);
            }
            queue.add(message);
            condition.signalAll();
        } finally {
            lock.unlock();
        }
    }

            synchronized (lock) {
                stopped = true;
            }
            stopped(this);
        }

        @Override
        public void dispatch(MethodInvocation message) {
            synchronized (lock) {
                if (stopped) {
                    throw new IllegalStateException("This actor has been stopped.");
                }
                dispatch.dispatch(message);