Malinskiy

/**
 * This type of request starts with single serialized request
 * and then proceed to do several reads and writes that have dynamic size
 */
abstract class ComplexRequest<T : Any?>(target: Target = NonSpecifiedTarget) : Request(target) {
    /**
     * Some requests ignore the initial OKAY/FAIL response and instead stream the actual response
     * To implement these we allow overriding this method
     */
    open suspend fun process(readChannel: AndroidReadChannel, writeChannel: AndroidWriteChannel): T {

Malinskiy

class AndroidReadChannel(private val delegate: ByteReadChannel) : ByteReadChannel by delegate {
    suspend fun read(): TransportResponse {
        val bytes = ByteArray(4)
        delegate.readFully(bytes, 0, 4)

        val ok = bytes.isOkay()
        val message = if (!ok) {
            delegate.readFully(bytes, 0, 4)
            val responseLength = String(bytes, Const.DEFAULT_TRANSPORT_ENCODING)
            val errorMessageLength = responseLength.toIntOrNull(16)

Malinskiy

abstract class Request(val target: Target = HostTarget) {

    /**
     * Some requests require a device serial to be passed to the request itself by means of <host-prefix>
     * @see https://android.googlesource.com/platform/system/core/+/refs/heads/master/adb/SERVICES.TXT
     */
    abstract fun serialize(): ByteArray

    open suspend fun handshake(readChannel: AndroidReadChannel, writeChannel: AndroidWriteChannel) {
        val request = serialize()

Malinskiy

sealed class Target {
    abstract fun serialize(): String
}
object HostTarget : Target() {
    override fun serialize() = "host:"
}
class SerialTarget(val serial: String) : Target() {
    override fun serialize() = "host-serial:$serial:"
}
object UsbTarget : Target() {

Malinskiy

suspend fun request() {
        aSocket(ActorSelectorManager())
            .tcp()
            .connect(InetSocketAddress("127.0.0.1", 5037)).use { socket ->
                //Work with the socket
            }
    }

Malinskiy

public class DefaultBuildScanEndOfBuildNotifier implements BuildScanEndOfBuildNotifier {
    private Listener listener;
    @Override
    public void notify(final Listener listener) {
        if (this.listener != null) {
            throw new IllegalStateException("Listener already registered");
        }
        this.listener = listener;
    }
    public void fireBuildComplete(@Nullable final Throwable failure) {

Malinskiy

open class SimpleMetric<T>(
  provider: (Unit) -> T, 
  assigner: (ExecutionReport, T) -> Unit) : Metric<T, Unit>(provider, assigner)

class ProcessorCountMetric : SimpleMetric<String>(
    provider = { Runtime.getRuntime().availableProcessors().toString() },
    assigner = { report, value -> report.environment.cpuCount = value }
)

Malinskiy

abstract class Metric<T, in Context>(
    val provider: (Context) -> T,
    val assigner: (ExecutionReport, T) -> Unit
) {
    open fun get(context: Context, report: ExecutionReport) {
        val value = provider(context)
        assigner(report, value)
        value
    }
}

Malinskiy

interface Tracker {
    fun trackTestTransition(poolId: DevicePoolId, transition: StateMachine.Transition<TestState, TestEvent, TestAction>)
    fun trackDeviceConnected(poolId: DevicePoolId, device: DeviceInfo)
    fun terminate()
}

Malinskiy

class DevicePoolActor(private val poolId: DevicePoolId,
                      private val configuration: Configuration,
                      analytics: Analytics,
                      tests: Collection<Test>,
                      private val progressReporter: ProgressReporter,
                      parent: Job,
                      context: CoroutineContext) :
        Actor<DevicePoolMessage>(parent = parent, context = context) {

    override suspend fun receive(msg: DevicePoolMessage) {