             let data = try! acceptedSocket.recv(1024)
             readData.append(String(bytes: data, encoding: NSUTF8StringEncoding)!)
             if readData.count >= 2 {
-                loop.removeReader(acceptedSocket.fileDescriptor)
+                self.loop.removeReader(acceptedSocket.fileDescriptor)
             }
         }
         
         loop.setReader(listenSocket.fileDescriptor) {
             acceptedSocket = try! listenSocket.accept()
-            loop.setReader(acceptedSocket.fileDescriptor, callback: readAcceptedSocket)
+            self.loop.setReader(acceptedSocket.fileDescriptor, callback: readAcceptedSocket)
         }
         
         try! clientSocket.connect("::1", port: port)
 
-        dispatch_after(dispatch_time(DISPATCH_TIME_NOW, Int64(1 * NSEC_PER_SEC)), queue) {
+        loop.callLater(1) {
             try! clientSocket.send(Array("hello".utf8))
         }
-        dispatch_after(dispatch_time(DISPATCH_TIME_NOW, Int64(2 * NSEC_PER_SEC)), queue) {
+        loop.callLater(2) {
             try! clientSocket.send(Array("baby".utf8))
         }
-        dispatch_after(dispatch_time(DISPATCH_TIME_NOW, Int64(3 * NSEC_PER_SEC)), queue) {
+        loop.callLater(3) {
             try! clientSocket.send(Array("fin".utf8))
         }
-        dispatch_after(dispatch_time(DISPATCH_TIME_NOW, Int64(4 * NSEC_PER_SEC)), queue) {
-            loop.stop()
+        loop.callLater(4) {
+            self.loop.stop()
         }
         
         assertExecutingTime(4.0, accuracy: 0.5) {
-            loop.runForever()
+            self.loop.runForever()
         }
         XCTAssertEqual(readData, ["hello", "baby"])
     }

            events.append(1)
        }
        loop.callLater(2) {
            loop.stop()
        }
        loop.callLater(3) {
            events.append(3)
        }
        assertExecutingTime(2, accuracy: 0.5) {
            loop.runForever()
        }
        XCTAssertEqual(events, [0, 1])
    }
    
    func testSetReader() {
        let loop = try! EventLoop(selector: try! KqueueSelector())
        
        let port = try! getUnusedTCPPort()
        let listenSocket = try! TCPSocket()
        try! listenSocket.bind(port)
        try! listenSocket.listen()
        
        loop.setReader(listenSocket.fileDescriptor) {
            loop.stop()
        }
        
        // make a connect 1 seconds later
        dispatch_after(dispatch_time(DISPATCH_TIME_NOW, Int64(1 * NSEC_PER_SEC)), queue) {
            let clientSocket = try! TCPSocket()
            try! clientSocket.connect("::1", port: port)
        }
        
        assertExecutingTime(1.0, accuracy: 0.5) {
            loop.runForever()
        }
    }
    
    func testSetWriter() {
        let loop = try! EventLoop(selector: try! KqueueSelector())
        
        let port = try! getUnusedTCPPort()
        let listenSocket = try! TCPSocket()
        try! listenSocket.bind(port)
        try! listenSocket.listen()
        
        let clientSocket = try! TCPSocket()
        
        loop.setWriter(clientSocket.fileDescriptor) {
            loop.stop()
        }
        
        // make a connect 1 seconds later
        dispatch_after(dispatch_time(DISPATCH_TIME_NOW, Int64(1 * NSEC_PER_SEC)), queue) {
            try! clientSocket.connect("::1", port: port)
        }
        
        assertExecutingTime(1.0, accuracy: 0.5) {
            loop.runForever()
        }
    }
    
    func testRemoveReader() {
        let loop = try! EventLoop(selector: try! KqueueSelector())
        
        let port = try! getUnusedTCPPort()
        let listenSocket = try! TCPSocket()
        try! listenSocket.bind(port)


class EventLoopTests: XCTestCase {
    let queue = dispatch_queue_create("com.envoy.embassy-tests.event-loop", DISPATCH_QUEUE_SERIAL)
    
    var loop: EventLoop!
    
    override func setUp() {
        super.setUp()
        loop = try! EventLoop(selector: try! KqueueSelector())
        
        // set a 30 seconds timeout
        dispatch_after(dispatch_time(DISPATCH_TIME_NOW, Int64(30 * NSEC_PER_SEC)), queue) {
            if self.loop.running {
                self.loop.stop()
                XCTFail("Time out")
            }
        }
    }
    
    override func tearDown() {
        super.tearDown()
    }
    
    func testStop() {
        dispatch_after(dispatch_time(DISPATCH_TIME_NOW, Int64(1 * NSEC_PER_SEC)), queue) {
            XCTAssert(self.loop.running)
            self.loop.stop()
            XCTAssertFalse(self.loop.running)
        }
        
        XCTAssertFalse(loop.running)
        assertExecutingTime(1.0, accuracy: 0.5) {
            self.loop.runForever()
        }
        XCTAssertFalse(loop.running)
    }
    
    func testCallSoon() {
        var called = false
        loop.callSoon {
            called = true
            self.loop.stop()
        }
        assertExecutingTime(0, accuracy: 0.5) {
            self.loop.runForever()
        }
        XCTAssert(called)
    }
    
    func testCallLater() {
        var events: [Int] = []
        loop.callLater(0) {
            events.append(0)
