}
    }

    private class TestConnector {
        private final BlockingQueue<InterHubMessage> incomingA = new LinkedBlockingQueue<>()
        private final BlockingQueue<InterHubMessage> outgoingA = new LinkedBlockingQueue<>()
        private final BlockingQueue<InterHubMessage> incomingB = new LinkedBlockingQueue<>()
        private final BlockingQueue<InterHubMessage> outgoingB = new LinkedBlockingQueue<>()

    InterHubMessage receive() {
        def message = incoming.take()
        return message == END ? null : (InterHubMessage) message
    }

    /**
     * Queues the given message to return from {@link #receive()}.
     */
    void queueIncoming(InterHubMessage message) {
        incoming.put(message)
    }

    /**
     * Marks the end of the incoming messages.
     */
    @Override
    void stop() {

    def "can access resolved files from afterResolve hook"() {
        given:
        file("thing.txt") << "stuff"
        buildFile << """
            configurations {
                things.incoming.afterResolve { incoming ->
                    incoming.files.files
                    println "accessed files"
                }
            }
            dependencies {
                things files("thing.txt")
            }

	m.blockQueue(func() {
		close(m.stopped)
		close(m.incoming)
	})
	m.distributing.Wait()
}

// loop receives from m.incoming and distributes to all watchers.
func (m *Broadcaster) loop() {
	// Deliberately not catching crashes here. Yes, bring down the process if there's a
	// bug in watch.Broadcaster.
	for event := range m.incoming {
		if event.Type == internalRunFunctionMarker {

        """
    }

    def "resolve the main classes through incoming.files"() {
        buildFile << """
            task verifyFiles(type: FilesVerificationTask) {
                incomingFiles.from(configurations.compileClasspath.incoming.files)
            }
        """
        expect:
        succeeds ':verifyFiles'
    }
    def "resolve the main classes through incoming.artifacts"() {
        buildFile << """

                def files2 = configurations.compile.incoming.artifactView({}).files
                def files3 = configurations.compile.incoming.artifactView({componentFilter { true }}).files
                def files4 = configurations.compile.incoming.artifactView({componentFilter { false }}).files
                def filterView = configurations.compile.incoming.artifactView({componentFilter(artifactFilter)})

                implementation 'com:foo:1.0:cls'
            }

            task resolve {
                def normal = configurations.runtimeClasspath.incoming.files
                def reselected = configurations.runtimeClasspath.incoming.artifactView {
                    withVariantReselection()
                    attributes {

    interface Worker {
        void doStuff(String value)
    }

    interface CompletableWorker extends Worker, StreamCompletion {
    }

    def "forwards incoming method invocations to handler"() {
        def worker = Mock(Worker)
        def connection = new TestConnection()

        given:
        connectCompletion.create(_) >> connection

        def outgoingMessage = new ChannelMessage(channel, "payload")

        given:
        def incoming = incomingQueue.getChannel(channel).newEndpoint()
        outgoingQueue.dispatch(outgoingMessage)

        when:
        connections.noFurtherConnections()
        def messages = []
        incoming.take(messages)

        then:
        messages.size() == 2
        messages[0] instanceof RejectedMessage

        println "artifacts 1: " + configurations.compile.incoming.artifacts.collect { it.file.name }
        println "artifacts 2: " + configurations.compile.incoming.artifactView { }.artifacts.collect { it.file.name }
        println "artifacts 3: " + configurations.compile.incoming.artifactView { lenient = true }.artifacts.collect { it.file.name }

        println "files 1: " + configurations.compile.incoming.files.collect { it.name }