Add queue as a parameter to PTUSBChannel connectToDevice

peertalk/PTChannel.m

@@ -197,7 +197,7 @@ - (void)connectToPort:(int)port overUSBHub:(PTUSBHub*)usbHub deviceID:(NSNumber*
     return;
   }
   connState_ = kConnStateConnecting;
-  [usbHub connectToDevice:deviceID port:port onStart:^(NSError *err, dispatch_io_t dispatchChannel) {
+  [usbHub connectToDevice:deviceID port:port onQueue:protocol_.queue onStart:^(NSError *err, dispatch_io_t dispatchChannel) {
     NSError *error = err;
     if (!error) {
       [self startReadingFromConnectedChannel:dispatchChannel error:&error];

peertalk/PTProtocol.m

@@ -307,7 +307,7 @@ - (void)readPayloadOfSize:(size_t)payloadSize overChannel:(dispatch_io_t)channel
 - (void)readAndDiscardDataOfSize:(size_t)size overChannel:(dispatch_io_t)channel callback:(void(^)(NSError*, BOOL))callback {
   dispatch_io_read(channel, 0, size, queue_, ^(bool done, dispatch_data_t data, int error) {
     if (done && callback) {
-      size_t dataSize = dispatch_data_get_size(data);
+      size_t dataSize = data ? dispatch_data_get_size(data) : 0;
       callback(error == 0 ? nil : [[NSError alloc] initWithDomain:NSPOSIXErrorDomain code:error userInfo:nil], dataSize == 0);
     }
   });

peertalk/PTUSBHub.h

@@ -61,6 +61,7 @@ typedef enum {
 //
 - (void)connectToDevice:(NSNumber*)deviceID
                    port:(int)port
+                onQueue:(dispatch_queue_t)queue
                 onStart:(void(^)(NSError *error, dispatch_io_t channel))onStart
                   onEnd:(void(^)(NSError *error))onEnd;
 

peertalk/PTUSBHub.m

@@ -219,11 +219,11 @@ - (void)listenOnQueue:(dispatch_queue_t)queue onStart:(void(^)(NSError*))onStart
 }
 
 
-- (void)connectToDevice:(NSNumber*)deviceID port:(int)port onStart:(void(^)(NSError*, dispatch_io_t))onStart onEnd:(void(^)(NSError*))onEnd {
+- (void)connectToDevice:(NSNumber*)deviceID port:(int)port onQueue:(dispatch_queue_t)queue onStart:(void(^)(NSError*, dispatch_io_t))onStart onEnd:(void(^)(NSError*))onEnd {
   PTUSBChannel *channel = [PTUSBChannel new];
   NSError *error = nil;
 
-  if (![channel openOnQueue:dispatch_get_main_queue() error:&error onEnd:onEnd]) {
+  if (![channel openOnQueue:queue error:&error onEnd:onEnd]) {
     onStart(error, nil);
     return;
   }
@@ -473,7 +473,8 @@ - (void)scheduleReadPacketWithBroadcastHandler:(void(^)(NSDictionary *packet))br
 - (void)scheduleReadPacketWithCallback:(void(^)(NSError*, NSDictionary*, uint32_t))callback {
   static usbmux_packet_t ref_upacket;
   isReadingPackets_ = YES;
-
+
+  // Read the first `sizeof(ref_upacket.size)` bytes off the channel_
   dispatch_io_read(channel_, 0, sizeof(ref_upacket.size), queue_, ^(bool done, dispatch_data_t data, int error) {
     //NSLog(@"dispatch_io_read 0,4: done=%d data=%p error=%d", done, data, error);
 
@@ -492,6 +493,7 @@ - (void)scheduleReadPacketWithCallback:(void(^)(NSError*, NSDictionary*, uint32_
     size_t buffer_size = 0;
     PT_PRECISE_LIFETIME_UNUSED dispatch_data_t map_data = dispatch_data_create_map(data, (const void **)&buffer, &buffer_size); // objc_precise_lifetime guarantees 'map_data' isn't released before memcpy has a chance to do its thing
     assert(buffer_size == sizeof(ref_upacket.size));
+    assert(sizeof(upacket_len) == sizeof(ref_upacket.size));
     memcpy((void *)&(upacket_len), (const void *)buffer, buffer_size);
 #if PT_DISPATCH_RETAIN_RELEASE
     dispatch_release(map_data);