CFStream.c

   1 /*
   2  * Copyright (c) 2014 Apple Inc. All rights reserved.
   3  *
   4  * @APPLE_LICENSE_HEADER_START@
   5  *
   6  * This file contains Original Code and/or Modifications of Original Code
   7  * as defined in and that are subject to the Apple Public Source License
   8  * Version 2.0 (the 'License'). You may not use this file except in
   9  * compliance with the License. Please obtain a copy of the License at
  10  * http://www.opensource.apple.com/apsl/ and read it before using this
  11  * file.
  12  *
  13  * The Original Code and all software distributed under the License are
  14  * distributed on an 'AS IS' basis, WITHOUT WARRANTY OF ANY KIND, EITHER
  15  * EXPRESS OR IMPLIED, AND APPLE HEREBY DISCLAIMS ALL SUCH WARRANTIES,
  16  * INCLUDING WITHOUT LIMITATION, ANY WARRANTIES OF MERCHANTABILITY,
  17  * FITNESS FOR A PARTICULAR PURPOSE, QUIET ENJOYMENT OR NON-INFRINGEMENT.
  18  * Please see the License for the specific language governing rights and
  19  * limitations under the License.
  20  *
  21  * @APPLE_LICENSE_HEADER_END@
  22  */
  23
  24 /*      CFStream.c
  25         Copyright (c) 2000-2013, Apple Inc. All rights reserved.
  26         Responsibility: John Iarocci
  27 */
  28
  29 #include <CoreFoundation/CFRuntime.h>
  30 #include <CoreFoundation/CFNumber.h>
  31 #include <string.h>
  32 #include "CFStreamInternal.h"
  33 #include "CFInternal.h"
  34 #include <stdio.h>
  35
  36 #if defined(DEBUG)
  37 #include <assert.h>
  38 #endif
  39
  40 struct _CFStream {
  41     CFRuntimeBase _cfBase;
  42     CFOptionFlags flags;
  43     CFErrorRef error; // if callBacks->version < 2, this is actually a pointer to a CFStreamError
  44     struct _CFStreamClient *client;
  45     /* NOTE: CFNetwork is still using _CFStreamGetInfoPointer, and so this slot needs to stay in this position (as the fifth field in the structure) */
  46     /* NOTE: This can be taken out once CFNetwork rebuilds */
  47     /* NOTE: <rdar://problem/13678879> Remove comment once CFNetwork has been rebuilt */
  48     void *info;
  49     const struct _CFStreamCallBacks *callBacks;  // This will not exist (will not be allocated) if the callbacks are from our known, "blessed" set.
  50
  51     CFSpinLock_t streamLock;
  52     CFArrayRef previousRunloopsAndModes;
  53     dispatch_queue_t queue;
  54 };
  55
  56
  57 enum {
  58         MIN_STATUS_CODE_BIT     = 0,
  59         // ..status bits...
  60         MAX_STATUS_CODE_BIT     = 4,
  61
  62         CONSTANT_CALLBACKS      = 5,
  63         CALLING_CLIENT          = 6,    // MUST remain 6 since it's value is used elsewhere.
  64
  65     HAVE_CLOSED                 = 7,
  66
  67     // Values above used to be defined and others may rely on their values
  68
  69     // Values below should not matter if they are re-ordered or shift
  70
  71     SHARED_SOURCE
  72 };
  73
  74
  75 /* CALLING_CLIENT really determines whether stream events will be sent to the client immediately, or posted for the next time through the runloop.  Since the client may not be prepared for re-entrancy, we must always set/clear this bit around public entry points. -- REW, 9/5/2001
  76     Also, CALLING_CLIENT is now used from CFFilteredStream.c (which has a copy of the #define above).  Really gross.  We should find a way to avoid that.... -- REW, 3/27/2002  */
  77 // Used in CFNetwork too
  78
  79 /* sSharesSources holds two mappings, one the inverse of the other, between a stream and the
  80    RunLoop+RunLoopMode pair that it's scheduled in.  If the stream is scheduled in more than
  81    one loop or mode, it can't share RunLoopSources with others, and is not in this dict.
  82 */
  83 static CFSpinLock_t sSourceLock = CFSpinLockInit;
  84 static CFMutableDictionaryRef sSharedSources = NULL;
  85
  86 static CFTypeID __kCFReadStreamTypeID = _kCFRuntimeNotATypeID;
  87 static CFTypeID __kCFWriteStreamTypeID = _kCFRuntimeNotATypeID;
  88
  89 // Just reads the bits, for those cases where we don't want to go through any callback checking
  90 #define __CFStreamGetStatus(x) __CFBitfieldGetValue((x)->flags, MAX_STATUS_CODE_BIT, MIN_STATUS_CODE_BIT)
  91
  92 CF_PRIVATE CFStreamStatus _CFStreamGetStatus(struct _CFStream *stream);
  93 static Boolean _CFStreamRemoveRunLoopAndModeFromArray(CFMutableArrayRef runLoopsAndModes, CFRunLoopRef rl, CFStringRef mode);
  94 static void _wakeUpRunLoop(struct _CFStream *stream);
  95
  96 CF_INLINE void checkRLMArray(CFArrayRef arr)
  97 {
  98 #if defined(DEBUG)
  99     assert(arr == NULL || (CFArrayGetCount(arr) % 2) == 0);
 100 #endif
 101 }
 102
 103 CF_INLINE void _CFStreamLock(struct _CFStream* stream) {
 104         __CFSpinLock(&stream->streamLock);
 105 }
 106
 107 CF_INLINE void _CFStreamUnlock(struct _CFStream* stream) {
 108         __CFSpinUnlock(&stream->streamLock);
 109 }
 110
 111 CF_INLINE CFRunLoopSourceRef _CFStreamCopySource(struct _CFStream* stream) {
 112     CFRunLoopSourceRef source = NULL;
 113
 114     if (stream) {
 115         _CFStreamLock(stream);
 116
 117         if (stream->client)
 118             source = stream->client->rlSource;
 119
 120         if (source)
 121             CFRetain(source);
 122
 123         _CFStreamUnlock(stream);
 124     }
 125
 126     return source;
 127 }
 128
 129 CF_INLINE void _CFStreamSetSource(struct _CFStream* stream, CFRunLoopSourceRef source, Boolean invalidateOldSource) {
 130         CFRunLoopSourceRef oldSource = NULL;
 131
 132         if (stream) {
 133                 _CFStreamLock(stream);
 134                 if (stream->client) {
 135                         oldSource = stream->client->rlSource;
 136                         if (oldSource != NULL)
 137                                 CFRetain(oldSource);
 138
 139                         stream->client->rlSource = source;
 140                         if (source != NULL)
 141                                 CFRetain(source);
 142                 }
 143                 _CFStreamUnlock(stream);
 144         }
 145
 146         if (oldSource) {
 147                 // Lose our extra retain
 148                 CFRelease(oldSource);
 149
 150                 if (invalidateOldSource)
 151                         CFRunLoopSourceInvalidate(oldSource);
 152
 153                 // And lose the one that held it in our stream as well
 154                 CFRelease(oldSource);
 155         }
 156 }
 157
 158 CF_INLINE const struct _CFStreamCallBacks *_CFStreamGetCallBackPtr(struct _CFStream *stream) {
 159     return stream->callBacks;
 160 }
 161
 162 CF_INLINE void _CFStreamSetStatusCode(struct _CFStream *stream, CFStreamStatus newStatus) {
 163     CFStreamStatus status = __CFStreamGetStatus(stream);
 164     if (((status != kCFStreamStatusClosed) && (status != kCFStreamStatusError)) ||
 165         ((status == kCFStreamStatusClosed) && (newStatus == kCFStreamStatusError)))
 166     {
 167         __CFBitfieldSetValue(stream->flags, MAX_STATUS_CODE_BIT, MIN_STATUS_CODE_BIT, newStatus);
 168     }
 169 }
 170
 171 CF_INLINE void _CFStreamScheduleEvent(struct _CFStream *stream, CFStreamEventType event) {
 172     if (stream->client && (stream->client->when & event)) {
 173                 CFRunLoopSourceRef source = _CFStreamCopySource(stream);
 174         if (source) {
 175             stream->client->whatToSignal |= event;
 176
 177                         CFRunLoopSourceSignal(source);
 178                         CFRelease(source);
 179                 _wakeUpRunLoop(stream);
 180             }
 181         }
 182 }
 183
 184 CF_INLINE void _CFStreamSetStreamError(struct _CFStream *stream, CFStreamError *err) {
 185     if (!stream->error) {
 186         stream->error = (CFErrorRef)CFAllocatorAllocate(CFGetAllocator(stream), sizeof(CFStreamError), 0);
 187     }
 188     memmove(stream->error, err, sizeof(CFStreamError));
 189 }
 190
 191 static CFStringRef __CFStreamCopyDescription(CFTypeRef cf) {
 192     struct _CFStream *stream = (struct _CFStream *)cf;
 193     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
 194     CFStringRef contextDescription;
 195     CFStringRef desc;
 196     if (cb->copyDescription) {
 197         if (cb->version == 0) {
 198             contextDescription = ((CFStringRef(*)(void *))cb->copyDescription)(_CFStreamGetInfoPointer(stream));
 199         } else {
 200             contextDescription = cb->copyDescription(stream, _CFStreamGetInfoPointer(stream));
 201         }
 202     } else {
 203         contextDescription = CFStringCreateWithFormat(CFGetAllocator(stream), NULL, CFSTR("info = %p"), _CFStreamGetInfoPointer(stream));
 204     }
 205     if (CFGetTypeID(cf) == __kCFReadStreamTypeID) {
 206         desc = CFStringCreateWithFormat(CFGetAllocator(stream), NULL, CFSTR("<CFReadStream %p>{%@}"), stream, contextDescription);
 207     } else {
 208         desc = CFStringCreateWithFormat(CFGetAllocator(stream), NULL, CFSTR("<CFWriteStream %p>{%@}"), stream, contextDescription);
 209     }
 210     CFRelease(contextDescription);
 211     return desc;
 212 }
 213
 214 static void _CFStreamDetachSource(struct _CFStream* stream) {
 215         if (stream && stream->client && stream->client->rlSource) {
 216         if (!__CFBitIsSet(stream->flags, SHARED_SOURCE)) {
 217                         _CFStreamSetSource(stream, NULL, TRUE);
 218         }
 219         else {
 220
 221             CFArrayRef runLoopAndSourceKey;
 222             CFMutableArrayRef list;
 223                         CFIndex count;
 224                         CFIndex i;
 225
 226             __CFSpinLock(&sSourceLock);
 227
 228             runLoopAndSourceKey = (CFArrayRef)CFDictionaryGetValue(sSharedSources, stream);
 229             list = (CFMutableArrayRef)CFDictionaryGetValue(sSharedSources, runLoopAndSourceKey);
 230
 231                         count = CFArrayGetCount(list);
 232                         i = CFArrayGetFirstIndexOfValue(list, CFRangeMake(0, count), stream);
 233             if (i != kCFNotFound) {
 234                 CFArrayRemoveValueAtIndex(list, i);
 235                                 count--;
 236             }
 237
 238             CFAssert(CFArrayGetFirstIndexOfValue(list, CFRangeMake(0, CFArrayGetCount(list)), stream) == kCFNotFound, __kCFLogAssertion, "CFStreamClose: stream found twice in its shared source's list");
 239
 240                         if (count == 0) {
 241                                 CFRunLoopSourceRef source = _CFStreamCopySource(stream);
 242                                 if (source) {
 243                                         CFRunLoopRemoveSource((CFRunLoopRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 0), source, (CFStringRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 1));
 244                                         CFRelease(source);
 245                                 }
 246                 CFDictionaryRemoveValue(sSharedSources, runLoopAndSourceKey);
 247             }
 248
 249             CFDictionaryRemoveValue(sSharedSources, stream);
 250
 251                         _CFStreamSetSource(stream, NULL, count == 0);
 252
 253             __CFBitClear(stream->flags, SHARED_SOURCE);
 254
 255             __CFSpinUnlock(&sSourceLock);
 256         }
 257     }
 258 }
 259
 260 CF_PRIVATE void _CFStreamClose(struct _CFStream *stream) {
 261     CFStreamStatus status = _CFStreamGetStatus(stream);
 262     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
 263     if (status == kCFStreamStatusNotOpen || status == kCFStreamStatusClosed || (status == kCFStreamStatusError && __CFBitIsSet(stream->flags, HAVE_CLOSED))) {
 264         // Stream is not open from the client's perspective; do not callout and do not update our status to "closed"
 265         return;
 266     }
 267     if (! __CFBitIsSet(stream->flags, HAVE_CLOSED)) {
 268         __CFBitSet(stream->flags, HAVE_CLOSED);
 269         __CFBitSet(stream->flags, CALLING_CLIENT);
 270         if (cb->close) {
 271             cb->close(stream, _CFStreamGetInfoPointer(stream));
 272         }
 273         if (stream->client) {
 274             _CFStreamDetachSource(stream);
 275         }
 276         _CFStreamSetStatusCode(stream, kCFStreamStatusClosed);
 277         __CFBitClear(stream->flags, CALLING_CLIENT);
 278     }
 279 }
 280
 281 //static int numStreamInstances = 0;
 282
 283 static void __CFStreamDeallocate(CFTypeRef cf) {
 284     struct _CFStream *stream = (struct _CFStream *)cf;
 285     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
 286     CFAllocatorRef alloc = CFGetAllocator(stream);
 287 //    numStreamInstances --;
 288
 289     // Close the stream
 290     _CFStreamClose(stream);
 291
 292     if (stream->client) {
 293         CFStreamClientContext *cbContext;
 294         cbContext = &(stream->client->cbContext);
 295         if (cbContext->info && cbContext->release) {
 296             cbContext->release(cbContext->info);
 297         }
 298
 299         if (stream->client->runLoopsAndModes) {
 300             CFRelease(stream->client->runLoopsAndModes);
 301         }
 302
 303         CFAllocatorDeallocate(alloc, stream->client);
 304         stream->client = NULL; // Just in case finalize, below, calls back in to us
 305     }
 306     if (cb->finalize) {
 307         if (cb->version == 0) {
 308             ((void(*)(void *))cb->finalize)(_CFStreamGetInfoPointer(stream));
 309         } else {
 310             cb->finalize(stream, _CFStreamGetInfoPointer(stream));
 311         }
 312     }
 313     if (stream->error) {
 314         if (cb->version < 2) {
 315             CFAllocatorDeallocate(alloc, stream->error);
 316         } else {
 317             CFRelease(stream->error);
 318         }
 319     }
 320     if (!__CFBitIsSet(stream->flags, CONSTANT_CALLBACKS)) {
 321         CFAllocatorDeallocate(alloc, (void *)stream->callBacks);
 322     }
 323     if (stream->previousRunloopsAndModes) {
 324         CFRelease(stream->previousRunloopsAndModes);
 325         stream->previousRunloopsAndModes = NULL;
 326     }
 327     if (stream->queue) {
 328         dispatch_release(stream->queue);
 329         stream->queue = NULL;
 330     }
 331 }
 332
 333 static const CFRuntimeClass __CFReadStreamClass = {
 334     0,
 335     "CFReadStream",
 336     NULL,      // init
 337     NULL,      // copy
 338     __CFStreamDeallocate,
 339     NULL,
 340     NULL,
 341     NULL,      // copyHumanDesc
 342     __CFStreamCopyDescription
 343 };
 344
 345 static const CFRuntimeClass __CFWriteStreamClass = {
 346     0,
 347     "CFWriteStream",
 348     NULL,      // init
 349     NULL,      // copy
 350     __CFStreamDeallocate,
 351     NULL,
 352     NULL,
 353     NULL,      // copyHumanDesc
 354     __CFStreamCopyDescription
 355 };
 356
 357 CONST_STRING_DECL(kCFStreamPropertySocketNativeHandle, "kCFStreamPropertySocketNativeHandle")
 358 CONST_STRING_DECL(kCFStreamPropertySocketRemoteHostName, "kCFStreamPropertySocketRemoteHostName")
 359 CONST_STRING_DECL(kCFStreamPropertySocketRemotePortNumber, "kCFStreamPropertySocketRemotePortNumber")
 360 CONST_STRING_DECL(kCFStreamPropertyDataWritten, "kCFStreamPropertyDataWritten")
 361 CONST_STRING_DECL(kCFStreamPropertyAppendToFile, "kCFStreamPropertyAppendToFile")
 362
 363 CF_PRIVATE void __CFStreamInitialize(void) {
 364     __kCFReadStreamTypeID = _CFRuntimeRegisterClass(&__CFReadStreamClass);
 365     __kCFWriteStreamTypeID = _CFRuntimeRegisterClass(&__CFWriteStreamClass);
 366 }
 367
 368
 369 CF_EXPORT CFTypeID CFReadStreamGetTypeID(void) {
 370     return __kCFReadStreamTypeID;
 371 }
 372
 373 CF_EXPORT CFTypeID CFWriteStreamGetTypeID(void) {
 374     return __kCFWriteStreamTypeID;
 375 }
 376
 377 static struct _CFStream *_CFStreamCreate(CFAllocatorRef allocator, Boolean isReadStream) {
 378     struct _CFStream *newStream = (struct _CFStream *)_CFRuntimeCreateInstance(allocator, isReadStream ? __kCFReadStreamTypeID : __kCFWriteStreamTypeID, sizeof(struct _CFStream) - sizeof(CFRuntimeBase), NULL);
 379     if (newStream) {
 380 //        numStreamInstances ++;
 381         newStream->flags = 0;
 382         _CFStreamSetStatusCode(newStream, kCFStreamStatusNotOpen);
 383         newStream->error = NULL;
 384         newStream->client = NULL;
 385         newStream->info = NULL;
 386         newStream->callBacks = NULL;
 387
 388         newStream->streamLock = CFSpinLockInit;
 389         newStream->previousRunloopsAndModes = NULL;
 390         newStream->queue = NULL;
 391     }
 392     return newStream;
 393 }
 394
 395 CF_EXPORT void* _CFStreamGetInfoPointer(struct _CFStream* stream) {
 396     return stream == NULL? NULL : stream->info;
 397 }
 398
 399 CF_PRIVATE struct _CFStream *_CFStreamCreateWithConstantCallbacks(CFAllocatorRef alloc, void *info,  const struct _CFStreamCallBacks *cb, Boolean isReading) {
 400     struct _CFStream *newStream;
 401     if (cb->version != 1) return NULL;
 402     newStream = _CFStreamCreate(alloc, isReading);
 403     if (newStream) {
 404         __CFBitSet(newStream->flags, CONSTANT_CALLBACKS);
 405         newStream->callBacks = cb;
 406         if (cb->create) {
 407             newStream->info = cb->create(newStream, info);
 408         } else {
 409             newStream->info = info;
 410         }
 411     }
 412     return newStream;
 413 }
 414
 415 CF_EXPORT void _CFStreamSetInfoPointer(struct _CFStream *stream, void *info, const struct _CFStreamCallBacks *cb) {
 416     if (info != stream->info) {
 417         if (stream->callBacks->finalize) {
 418             stream->callBacks->finalize(stream, stream->info);
 419         }
 420         if (cb->create) {
 421             stream->info = cb->create(stream, info);
 422         } else {
 423             stream->info = info;
 424         }
 425     }
 426     stream->callBacks = cb;
 427 }
 428
 429
 430 CF_EXPORT CFReadStreamRef CFReadStreamCreate(CFAllocatorRef alloc, const CFReadStreamCallBacks *callbacks, void *info) {
 431     struct _CFStream *newStream = _CFStreamCreate(alloc, TRUE);
 432     struct _CFStreamCallBacks *cb;
 433     if (!newStream) return NULL;
 434     cb = (struct _CFStreamCallBacks *)CFAllocatorAllocate(alloc, sizeof(struct _CFStreamCallBacks), 0);
 435     if (!cb) {
 436         CFRelease(newStream);
 437         return NULL;
 438     }
 439     if (callbacks->version == 0) {
 440         CFReadStreamCallBacksV0 *cbV0 = (CFReadStreamCallBacksV0 *)callbacks;
 441         CFStreamClientContext *ctxt = (CFStreamClientContext *)info;
 442         newStream->info = ctxt->retain ? (void *)ctxt->retain(ctxt->info) : ctxt->info;
 443         cb->version = 0;
 444         cb->create = (void *(*)(struct _CFStream *, void *))ctxt->retain;
 445         cb->finalize = (void(*)(struct _CFStream *, void *))ctxt->release;
 446         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))ctxt->copyDescription;
 447         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))cbV0->open;
 448         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))cbV0->openCompleted;
 449         cb->read = (CFIndex (*)(CFReadStreamRef, UInt8 *, CFIndex, CFErrorRef *, Boolean *, void *))cbV0->read;
 450         cb->getBuffer = (const UInt8 *(*)(CFReadStreamRef, CFIndex, CFIndex *, CFErrorRef *, Boolean *, void *))cbV0->getBuffer;
 451         cb->canRead = (Boolean (*)(CFReadStreamRef, CFErrorRef*, void*))cbV0->canRead;
 452         cb->write = NULL;
 453         cb->canWrite = NULL;
 454         cb->close = (void (*)(struct _CFStream *, void *))cbV0->close;
 455         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))cbV0->copyProperty;
 456         cb->setProperty = NULL;
 457         cb->requestEvents = NULL;
 458         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV0->schedule;
 459         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV0->unschedule;
 460     } else if (callbacks->version == 1) {
 461         CFReadStreamCallBacksV1 *cbV1 = (CFReadStreamCallBacksV1 *)callbacks;
 462         newStream->info = cbV1->create ? cbV1->create((CFReadStreamRef)newStream, info) : info;
 463         cb->version = 1;
 464         cb->create = (void *(*)(struct _CFStream *, void *))cbV1->create;
 465         cb->finalize = (void(*)(struct _CFStream *, void *))cbV1->finalize;
 466         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))cbV1->copyDescription;
 467         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))cbV1->open;
 468         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))cbV1->openCompleted;
 469         cb->read = (CFIndex (*)(CFReadStreamRef, UInt8 *, CFIndex, CFErrorRef *, Boolean *, void *))cbV1->read;
 470         cb->getBuffer = (const UInt8 *(*)(CFReadStreamRef, CFIndex, CFIndex *, CFErrorRef *, Boolean *, void *))cbV1->getBuffer;
 471         cb->canRead = (Boolean (*)(CFReadStreamRef, CFErrorRef*, void*))cbV1->canRead;
 472         cb->write = NULL;
 473         cb->canWrite = NULL;
 474         cb->close = (void (*)(struct _CFStream *, void *))cbV1->close;
 475         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))cbV1->copyProperty;
 476         cb->setProperty = (Boolean(*)(struct _CFStream *, CFStringRef, CFTypeRef, void *))cbV1->setProperty;
 477         cb->requestEvents = (void(*)(struct _CFStream *, CFOptionFlags, void *))cbV1->requestEvents;
 478         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV1->schedule;
 479         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV1->unschedule;
 480     } else {
 481         newStream->info = callbacks->create ? callbacks->create((CFReadStreamRef)newStream, info) : info;
 482         cb->version = 2;
 483         cb->create = (void *(*)(struct _CFStream *, void *))callbacks->create;
 484         cb->finalize = (void(*)(struct _CFStream *, void *))callbacks->finalize;
 485         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))callbacks->copyDescription;
 486         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))callbacks->open;
 487         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))callbacks->openCompleted;
 488         cb->read = callbacks->read;
 489         cb->getBuffer = callbacks->getBuffer;
 490         cb->canRead = callbacks->canRead;
 491         cb->write = NULL;
 492         cb->canWrite = NULL;
 493         cb->close = (void (*)(struct _CFStream *, void *))callbacks->close;
 494         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))callbacks->copyProperty;
 495         cb->setProperty = (Boolean(*)(struct _CFStream *, CFStringRef, CFTypeRef, void *))callbacks->setProperty;
 496         cb->requestEvents = (void(*)(struct _CFStream *, CFOptionFlags, void *))callbacks->requestEvents;
 497         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))callbacks->schedule;
 498         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))callbacks->unschedule;
 499    }
 500
 501     newStream->callBacks = cb;
 502     return (CFReadStreamRef)newStream;
 503 }
 504
 505 CF_EXPORT CFWriteStreamRef CFWriteStreamCreate(CFAllocatorRef alloc, const CFWriteStreamCallBacks *callbacks, void *info) {
 506     struct _CFStream *newStream = _CFStreamCreate(alloc, FALSE);
 507     struct _CFStreamCallBacks *cb;
 508     if (!newStream) return NULL;
 509     cb = (struct _CFStreamCallBacks *)CFAllocatorAllocate(alloc, sizeof(struct _CFStreamCallBacks), 0);
 510     if (!cb) {
 511         CFRelease(newStream);
 512         return NULL;
 513     }
 514     if (callbacks->version == 0) {
 515         CFWriteStreamCallBacksV0 *cbV0 = (CFWriteStreamCallBacksV0 *)callbacks;
 516         CFStreamClientContext *ctxt = (CFStreamClientContext *)info;
 517         newStream->info = ctxt->retain ? (void *)ctxt->retain(ctxt->info) : ctxt->info;
 518         cb->version = 0;
 519         cb->create = (void *(*)(struct _CFStream *, void *))ctxt->retain;
 520         cb->finalize = (void(*)(struct _CFStream *, void *))ctxt->release;
 521         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))ctxt->copyDescription;
 522         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))cbV0->open;
 523         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))cbV0->openCompleted;
 524         cb->read = NULL;
 525         cb->getBuffer = NULL;
 526         cb->canRead = NULL;
 527         cb->write = (CFIndex(*)(CFWriteStreamRef stream, const UInt8 *buffer, CFIndex bufferLength, CFErrorRef *error, void *info))cbV0->write;
 528         cb->canWrite = (Boolean(*)(CFWriteStreamRef stream, CFErrorRef *error, void *info))cbV0->canWrite;
 529         cb->close = (void (*)(struct _CFStream *, void *))cbV0->close;
 530         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))cbV0->copyProperty;
 531         cb->setProperty = NULL;
 532         cb->requestEvents = NULL;
 533         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV0->schedule;
 534         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV0->unschedule;
 535     } else if (callbacks->version == 1) {
 536         CFWriteStreamCallBacksV1 *cbV1 = (CFWriteStreamCallBacksV1 *)callbacks;
 537         cb->version = 1;
 538         newStream->info = cbV1->create ? cbV1->create((CFWriteStreamRef)newStream, info) : info;
 539         cb->create = (void *(*)(struct _CFStream *, void *))cbV1->create;
 540         cb->finalize = (void(*)(struct _CFStream *, void *))cbV1->finalize;
 541         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))cbV1->copyDescription;
 542         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))cbV1->open;
 543         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))cbV1->openCompleted;
 544         cb->read = NULL;
 545         cb->getBuffer = NULL;
 546         cb->canRead = NULL;
 547         cb->write = (CFIndex(*)(CFWriteStreamRef stream, const UInt8 *buffer, CFIndex bufferLength, CFErrorRef *error, void *info))cbV1->write;
 548         cb->canWrite = (Boolean(*)(CFWriteStreamRef stream, CFErrorRef *error, void *info))cbV1->canWrite;
 549         cb->close = (void (*)(struct _CFStream *, void *))cbV1->close;
 550         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))cbV1->copyProperty;
 551         cb->setProperty = (Boolean (*)(struct _CFStream *, CFStringRef, CFTypeRef, void *))cbV1->setProperty;
 552         cb->requestEvents = (void(*)(struct _CFStream *, CFOptionFlags, void *))cbV1->requestEvents;
 553         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV1->schedule;
 554         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV1->unschedule;
 555     } else {
 556         cb->version = callbacks->version;
 557         newStream->info = callbacks->create ? callbacks->create((CFWriteStreamRef)newStream, info) : info;
 558         cb->create = (void *(*)(struct _CFStream *, void *))callbacks->create;
 559         cb->finalize = (void(*)(struct _CFStream *, void *))callbacks->finalize;
 560         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))callbacks->copyDescription;
 561         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))callbacks->open;
 562         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))callbacks->openCompleted;
 563         cb->read = NULL;
 564         cb->getBuffer = NULL;
 565         cb->canRead = NULL;
 566         cb->write = callbacks->write;
 567         cb->canWrite = callbacks->canWrite;
 568         cb->close = (void (*)(struct _CFStream *, void *))callbacks->close;
 569         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))callbacks->copyProperty;
 570         cb->setProperty = (Boolean (*)(struct _CFStream *, CFStringRef, CFTypeRef, void *))callbacks->setProperty;
 571         cb->requestEvents = (void(*)(struct _CFStream *, CFOptionFlags, void *))callbacks->requestEvents;
 572         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))callbacks->schedule;
 573         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))callbacks->unschedule;
 574     }
 575     newStream->callBacks = cb;
 576     return (CFWriteStreamRef)newStream;
 577 }
 578
 579 static void _signalEventSync(struct _CFStream* stream, CFOptionFlags whatToSignal)
 580 {
 581     CFOptionFlags eventMask;
 582
 583     __CFBitSet(stream->flags, CALLING_CLIENT);
 584
 585     _CFStreamLock(stream);
 586
 587     struct _CFStreamClient* client = stream->client;
 588     if (client == NULL) {
 589         _CFStreamUnlock(stream);
 590     } else {
 591         void* info = NULL;
 592         void (*release) (void*) = NULL;
 593         void (*cb)(struct _CFStream *, CFStreamEventType, void *) = client == NULL? NULL : client->cb;
 594
 595         if (stream->client->cbContext.retain == NULL)
 596             info = stream->client->cbContext.info;
 597         else {
 598             info = stream->client->cbContext.retain(stream->client->cbContext.info);
 599             release = stream->client->cbContext.release;
 600         }
 601         _CFStreamUnlock(stream);
 602
 603         for (eventMask = 1; eventMask <= whatToSignal; eventMask = eventMask << 1) {
 604             _CFStreamLock(stream);
 605             Boolean shouldSignal = ((eventMask & whatToSignal) && stream->client && (stream->client->when & eventMask));
 606             _CFStreamUnlock(stream);
 607
 608             if (shouldSignal && client) {
 609                 cb(stream, eventMask, info);
 610
 611                 /* What happens if the callback sets the client to NULL?  We're in a loop here... Hmm. */
 612                 /* After writing that comment, I see: <rdar://problem/6793636> CFReadStreamSetClient(..., NULL) unsafely releases info pointer immediately */
 613                 /* Of note, when the stream callbacks are set to to NULL, we're re-initalized so as not to receive more events, so we
 614                  * should break pout of this loop */
 615             }
 616         }
 617
 618         if (release)
 619             (*release) (info);
 620     }
 621     __CFBitClear(stream->flags, CALLING_CLIENT);
 622 }
 623
 624 static void _signalEventQueue(dispatch_queue_t q, struct _CFStream* stream, CFOptionFlags whatToSignal)
 625 {
 626     CFRetain(stream);
 627     dispatch_async(q, ^{
 628         _signalEventSync(stream, whatToSignal);
 629         CFRelease(stream);
 630     });
 631 }
 632
 633 static void _cfstream_solo_signalEventSync(void* info)
 634 {
 635     CFTypeID typeID = CFGetTypeID((CFTypeRef) info);
 636
 637     if (typeID != CFReadStreamGetTypeID() && typeID != CFWriteStreamGetTypeID()) {
 638         CFLog(__kCFLogAssertion, CFSTR("Expected a read or write stream for %p"), info);
 639 #if defined(DEBUG)
 640         abort();
 641 #endif
 642     } else {
 643         struct _CFStream* stream = (struct _CFStream*) info;
 644         _CFStreamLock(stream);
 645         CFOptionFlags whatToSignal = stream->client->whatToSignal;
 646         stream->client->whatToSignal = 0;
 647         dispatch_queue_t queue = stream->queue;
 648         if (queue) dispatch_retain(queue);
 649         CFRetain(stream);
 650         _CFStreamUnlock(stream);
 651
 652         /* Since the array version holds a retain, we do it here as well, as opposed to taking a second retain in the client callback */
 653         if (queue == 0)
 654             _signalEventSync(stream, whatToSignal);
 655         else {
 656             _signalEventQueue(queue, stream, whatToSignal);
 657             dispatch_release(queue);
 658         }
 659         CFRelease(stream);
 660     }
 661 }
 662
 663 static void _cfstream_shared_signalEventSync(void* info)
 664 {
 665     CFTypeID typeID = CFGetTypeID((CFTypeRef) info);
 666
 667     if (typeID != CFArrayGetTypeID()) {
 668         CFLog(__kCFLogAssertion, CFSTR("Expected an array for %p"), info);
 669 #if defined(DEBUG)
 670         abort();
 671 #endif
 672     } else {
 673         CFMutableArrayRef list = (CFMutableArrayRef) info;
 674         CFIndex c, i;
 675         CFOptionFlags whatToSignal = 0;
 676         dispatch_queue_t queue = 0;
 677         struct _CFStream* stream = NULL;
 678
 679         __CFSpinLock(&sSourceLock);
 680
 681         /* Looks like, we grab the first stream that wants an event... */
 682         /* Note that I grab an extra retain when I pull out the stream here... */
 683         c = CFArrayGetCount(list);
 684         for (i = 0; i < c; i++) {
 685             struct _CFStream* s = (struct _CFStream*)CFArrayGetValueAtIndex(list, i);
 686
 687             if (s->client->whatToSignal) {
 688                 stream = s;
 689                 CFRetain(stream);
 690                 whatToSignal = stream->client->whatToSignal;
 691                 s->client->whatToSignal = 0;
 692                 queue = stream->queue;
 693                 if (queue) dispatch_retain(queue);
 694                 break;
 695             }
 696         }
 697
 698         /* And then we also signal any other streams in this array so that we get them next go? */
 699         for (; i < c;  i++) {
 700             struct _CFStream* s = (struct _CFStream*)CFArrayGetValueAtIndex(list, i);
 701             if (s->client->whatToSignal) {
 702                 CFRunLoopSourceRef source = _CFStreamCopySource(s);
 703                 if (source) {
 704                     CFRunLoopSourceSignal(source);
 705                     CFRelease(source);
 706                 }
 707                 break;
 708             }
 709         }
 710
 711         __CFSpinUnlock(&sSourceLock);
 712
 713         /* We're sitting here now, possibly with a stream that needs to be processed by the common routine */
 714         if (stream) {
 715             if (queue == 0)
 716                 _signalEventSync(stream, whatToSignal);
 717             else {
 718                 _signalEventQueue(queue, stream, whatToSignal);
 719                 dispatch_release(queue);
 720             }
 721
 722             /* Lose our extra retain */
 723             CFRelease(stream);
 724         }
 725     }
 726 }
 727
 728 /* This routine is to be considered unsafe... */
 729 static CFArrayRef _CFStreamGetRunLoopsAndModes(struct _CFStream *stream)
 730 {
 731     CFArrayRef result = NULL;
 732     if (stream && stream->client) {
 733         _CFStreamLock(stream);
 734         if (stream->previousRunloopsAndModes) {
 735             CFRelease(stream->previousRunloopsAndModes);
 736             stream->previousRunloopsAndModes = NULL;
 737         }
 738         if (stream->client->runLoopsAndModes) {
 739             stream->previousRunloopsAndModes = CFArrayCreateCopy(CFGetAllocator(stream), stream->client->runLoopsAndModes);
 740         }
 741         result = stream->previousRunloopsAndModes;
 742         checkRLMArray(result);
 743         _CFStreamUnlock(stream);
 744     }
 745     return result;
 746 }
 747
 748 static CFArrayRef _CFStreamCopyRunLoopsAndModes(struct _CFStream *stream)
 749 {
 750     CFArrayRef result = NULL;
 751     if (stream && stream->client) {
 752         _CFStreamLock(stream);
 753         if (stream->client->runLoopsAndModes) {
 754             result = CFArrayCreateCopy(CFGetAllocator(stream), stream->client->runLoopsAndModes);
 755         }
 756         checkRLMArray(result);
 757         _CFStreamUnlock(stream);
 758     }
 759     return result;
 760 }
 761
 762 static void _wakeUpRunLoop(struct _CFStream *stream) {
 763     CFArrayRef rlArray = _CFStreamCopyRunLoopsAndModes(stream);
 764     if (rlArray) {
 765         CFIndex cnt = CFArrayGetCount(rlArray);
 766         CFRunLoopRef rl = NULL;
 767
 768         if (cnt == 2) {
 769             rl = (CFRunLoopRef)CFArrayGetValueAtIndex(rlArray, 0);
 770         } else if (cnt > 2) {
 771             CFIndex idx;
 772             rl = (CFRunLoopRef)CFArrayGetValueAtIndex(rlArray, 0);
 773             for (idx = 2; NULL != rl && idx < cnt; idx+=2) {
 774                 CFRunLoopRef value = (CFRunLoopRef)CFArrayGetValueAtIndex(rlArray, idx);
 775                 if (value != rl) rl = NULL;
 776             }
 777             if (NULL == rl) {   /* more than one different rl, so we must pick one */
 778                 for (idx = 0; idx < cnt; idx+=2) {
 779                     CFRunLoopRef value = (CFRunLoopRef)CFArrayGetValueAtIndex(rlArray, idx);
 780                     CFStringRef currentMode = CFRunLoopCopyCurrentMode(value);
 781                     if (NULL != currentMode && CFEqual(currentMode, CFArrayGetValueAtIndex(rlArray, idx+1)) && CFRunLoopIsWaiting(value)) {
 782                         CFRelease(currentMode);
 783                         rl = value;
 784                         break;
 785                     }
 786                     if (NULL != currentMode) CFRelease(currentMode);
 787                 }
 788                 if (NULL == rl) {       /* didn't choose one above, so choose first */
 789                     rl = (CFRunLoopRef)CFArrayGetValueAtIndex(rlArray, 0);
 790                 }
 791             }
 792         }
 793         if (NULL != rl && CFRunLoopIsWaiting(rl))
 794             CFRunLoopWakeUp(rl);
 795         CFRelease(rlArray);
 796     }
 797 }
 798
 799 CF_PRIVATE void _CFStreamSignalEvent(struct _CFStream *stream, CFStreamEventType event, CFErrorRef error, Boolean synchronousAllowed) {
 800     // Update our internal status; we must use the primitive __CFStreamGetStatus(), because CFStreamGetStatus() calls us, and we can end up in an infinite loop.
 801     CFStreamStatus status = __CFStreamGetStatus(stream);
 802
 803     // Sanity check the event
 804     if (status == kCFStreamStatusNotOpen) {
 805         // No events allowed; this is almost certainly a bug in the stream's implementation
 806         CFLog(__kCFLogAssertion, CFSTR("Stream %p is sending an event before being opened"), stream);
 807         event = 0;
 808     } else if (status == kCFStreamStatusClosed || status == kCFStreamStatusError) {
 809         // no further events are allowed
 810         event = 0;
 811     } else if (status == kCFStreamStatusAtEnd) {
 812         // Only error events are allowed
 813         event &= kCFStreamEventErrorOccurred;
 814     } else if (status != kCFStreamStatusOpening) {
 815         // cannot send open completed; that happened already
 816         event &= ~kCFStreamEventOpenCompleted;
 817     }
 818
 819     // Change status if appropriate
 820     if (event & kCFStreamEventOpenCompleted && status == kCFStreamStatusOpening) {
 821         _CFStreamSetStatusCode(stream, kCFStreamStatusOpen);
 822     }
 823     if (event & kCFStreamEventEndEncountered && status < kCFStreamStatusAtEnd) {
 824         _CFStreamSetStatusCode(stream, kCFStreamStatusAtEnd);
 825     }
 826     if (event & kCFStreamEventErrorOccurred) {
 827         if (_CFStreamGetCallBackPtr(stream)->version < 2) {
 828             _CFStreamSetStreamError(stream, (CFStreamError *)error);
 829         } else {
 830             CFAssert(error, __kCFLogAssertion, "CFStream: kCFStreamEventErrorOccurred signalled, but error is NULL!");
 831             CFRetain(error);
 832             if (stream->error) CFRelease(stream->error);
 833             stream->error = error;
 834         }
 835         _CFStreamSetStatusCode(stream, kCFStreamStatusError);
 836     }
 837
 838     // Now signal any pertinent event
 839     if (stream->client && (stream->client->when & event) != 0) {
 840         CFRunLoopSourceRef source = _CFStreamCopySource(stream);
 841
 842         if (source) {
 843             Boolean signalNow = FALSE;
 844
 845             stream->client->whatToSignal |= event;
 846
 847             if (synchronousAllowed && !__CFBitIsSet(stream->flags, CALLING_CLIENT)) {
 848
 849                 CFRunLoopRef rl = CFRunLoopGetCurrent();
 850                 CFStringRef mode = CFRunLoopCopyCurrentMode(rl);
 851
 852                 if (mode) {
 853                     if (CFRunLoopContainsSource(rl, source, mode))
 854                         signalNow = TRUE;
 855                 }
 856                 if (mode)
 857                     CFRelease(mode);
 858             }
 859
 860             if (signalNow) {
 861                 // Can call out safely right now
 862                 _cfstream_solo_signalEventSync(stream);
 863             } else {
 864                 // Schedule for later delivery
 865                 if (source) {
 866                     CFRunLoopSourceSignal(source);
 867                 }
 868                 _wakeUpRunLoop(stream);
 869             }
 870
 871             CFRelease(source);
 872         }
 873     }
 874 }
 875
 876 CF_PRIVATE CFStreamStatus _CFStreamGetStatus(struct _CFStream *stream) {
 877   CFStreamStatus status = __CFStreamGetStatus(stream);
 878   // Status code just represents the value when last we checked; if we were in the middle of doing work at that time, we need to find out if the work has completed, now.  If we find out about a status change, we need to inform the client as well, so we call _CFStreamSignalEvent.  This will take care of updating our internal status correctly, too.
 879   __CFBitSet(stream->flags, CALLING_CLIENT);
 880   if (status == kCFStreamStatusOpening) {
 881     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
 882     if (cb->openCompleted) {
 883       Boolean isComplete;
 884       if (cb->version < 2) {
 885         CFStreamError err = {0, 0};
 886         isComplete = ((_CFStreamCBOpenCompletedV1)(cb->openCompleted))(stream, &err, _CFStreamGetInfoPointer(stream));
 887         if (err.error != 0) _CFStreamSetStreamError(stream, &err);
 888       } else {
 889         isComplete = cb->openCompleted(stream, &(stream->error), _CFStreamGetInfoPointer(stream));
 890       }
 891       if (isComplete) {
 892         if (!stream->error) {
 893           status = kCFStreamStatusOpen;
 894         } else {
 895           status = kCFStreamStatusError;
 896         }
 897         _CFStreamSetStatusCode(stream, status);
 898         if (status == kCFStreamStatusOpen) {
 899           _CFStreamScheduleEvent(stream, kCFStreamEventOpenCompleted);
 900         } else {
 901           _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
 902         }
 903       }
 904     }
 905   }
 906   __CFBitClear(stream->flags, CALLING_CLIENT);
 907   return status;
 908 }
 909
 910 CF_EXPORT CFStreamStatus CFReadStreamGetStatus(CFReadStreamRef stream) {
 911     CF_OBJC_FUNCDISPATCHV(__kCFReadStreamTypeID, CFStreamStatus, (NSInputStream *)stream, streamStatus);
 912     return _CFStreamGetStatus((struct _CFStream *)stream);
 913 }
 914
 915 CF_EXPORT CFStreamStatus CFWriteStreamGetStatus(CFWriteStreamRef stream) {
 916     CF_OBJC_FUNCDISPATCHV(__kCFWriteStreamTypeID, CFStreamStatus, (NSOutputStream *)stream, streamStatus);
 917     return _CFStreamGetStatus((struct _CFStream *)stream);
 918 }
 919
 920 static CFStreamError _CFStreamGetStreamError(struct _CFStream *stream) {
 921     CFStreamError result;
 922     if (!stream->error) {
 923         result.error = 0;
 924         result.domain = 0;
 925     } else if (_CFStreamGetCallBackPtr(stream)->version < 2) {
 926         CFStreamError *streamError = (CFStreamError *)(stream->error);
 927         result.error = streamError->error;
 928         result.domain = streamError->domain;
 929     } else {
 930         result = _CFStreamErrorFromError(stream->error);
 931     }
 932     return result;
 933 }
 934
 935 CF_EXPORT CFStreamError CFReadStreamGetError(CFReadStreamRef stream) {
 936     CF_OBJC_FUNCDISPATCHV(__kCFReadStreamTypeID, CFStreamError, (NSInputStream *)stream, _cfStreamError);
 937     return _CFStreamGetStreamError((struct _CFStream *)stream);
 938 }
 939
 940 CF_EXPORT CFStreamError CFWriteStreamGetError(CFWriteStreamRef stream) {
 941     CF_OBJC_FUNCDISPATCHV(__kCFWriteStreamTypeID, CFStreamError, (NSOutputStream *)stream, _cfStreamError);
 942     return _CFStreamGetStreamError((struct _CFStream *)stream);
 943 }
 944
 945 static CFErrorRef _CFStreamCopyError(struct _CFStream *stream) {
 946     if (!stream->error) {
 947         return NULL;
 948     } else if (_CFStreamGetCallBackPtr(stream)->version < 2) {
 949         return _CFErrorFromStreamError(CFGetAllocator(stream), (CFStreamError *)(stream->error));
 950     } else {
 951         CFRetain(stream->error);
 952         return stream->error;
 953     }
 954 }
 955
 956 CF_EXPORT CFErrorRef CFReadStreamCopyError(CFReadStreamRef stream) {
 957     CF_OBJC_FUNCDISPATCHV(__kCFReadStreamTypeID, CFErrorRef, (NSInputStream *)stream, streamError);
 958     return _CFStreamCopyError((struct _CFStream *)stream);
 959 }
 960
 961 CF_EXPORT CFErrorRef CFWriteStreamCopyError(CFWriteStreamRef stream) {
 962     return _CFStreamCopyError((struct _CFStream *)stream);
 963     CF_OBJC_FUNCDISPATCHV(__kCFWriteStreamTypeID, CFErrorRef, (NSOutputStream *)stream, streamError);
 964 }
 965
 966 CF_PRIVATE Boolean _CFStreamOpen(struct _CFStream *stream) {
 967     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
 968     Boolean success, openComplete;
 969     if (_CFStreamGetStatus(stream) != kCFStreamStatusNotOpen) {
 970         return FALSE;
 971     }
 972     __CFBitSet(stream->flags, CALLING_CLIENT);
 973     _CFStreamSetStatusCode(stream, kCFStreamStatusOpening);
 974     if (cb->open) {
 975         if (cb->version < 2) {
 976             CFStreamError err = {0, 0};
 977             success = ((_CFStreamCBOpenV1)(cb->open))(stream, &err, &openComplete, _CFStreamGetInfoPointer(stream));
 978             if (err.error != 0) _CFStreamSetStreamError(stream, &err);
 979         } else {
 980             success = cb->open(stream, &(stream->error), &openComplete, _CFStreamGetInfoPointer(stream));
 981         }
 982     } else {
 983         success = TRUE;
 984         openComplete = TRUE;
 985     }
 986     if (openComplete) {
 987         if (success) {
 988             // 2957690 - Guard against the possibility that the stream has already signalled itself in to a later state (like AtEnd)
 989             if (__CFStreamGetStatus(stream) == kCFStreamStatusOpening) {
 990                 _CFStreamSetStatusCode(stream, kCFStreamStatusOpen);
 991             }
 992             _CFStreamScheduleEvent(stream, kCFStreamEventOpenCompleted);
 993         } else {
 994 #if DEPLOYMENT_TARGET_WINDOWS
 995             _CFStreamClose(stream);
 996 #endif
 997             _CFStreamSetStatusCode(stream, kCFStreamStatusError);
 998             _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
 999         }
1000     }
1001     __CFBitClear(stream->flags, CALLING_CLIENT);
1002     return success;
1003 }
1004
1005 CF_EXPORT Boolean CFReadStreamOpen(CFReadStreamRef stream) {
1006     if(CF_IS_OBJC(__kCFReadStreamTypeID, stream)) {
1007         (void)CF_OBJC_CALLV((NSInputStream *)stream, open);
1008         return TRUE;
1009     }
1010     return _CFStreamOpen((struct _CFStream *)stream);
1011 }
1012
1013 CF_EXPORT Boolean CFWriteStreamOpen(CFWriteStreamRef stream) {
1014     if(CF_IS_OBJC(__kCFWriteStreamTypeID, stream)) {
1015         (void)CF_OBJC_CALLV((NSOutputStream *)stream, open);
1016         return TRUE;
1017     }
1018     return _CFStreamOpen((struct _CFStream *)stream);
1019 }
1020
1021 CF_EXPORT void CFReadStreamClose(CFReadStreamRef stream) {
1022     CF_OBJC_FUNCDISPATCHV(__kCFReadStreamTypeID, void, (NSInputStream *)stream, close);
1023     _CFStreamClose((struct _CFStream *)stream);
1024 }
1025
1026 CF_EXPORT void CFWriteStreamClose(CFWriteStreamRef stream) {
1027     CF_OBJC_FUNCDISPATCHV(__kCFWriteStreamTypeID, void, (NSOutputStream *)stream, close);
1028     _CFStreamClose((struct _CFStream *)stream);
1029 }
1030
1031 CF_EXPORT Boolean CFReadStreamHasBytesAvailable(CFReadStreamRef readStream) {
1032     CF_OBJC_FUNCDISPATCHV(__kCFReadStreamTypeID, Boolean, (NSInputStream *)readStream, hasBytesAvailable);
1033     struct _CFStream *stream = (struct _CFStream *)readStream;
1034     CFStreamStatus status = _CFStreamGetStatus(stream);
1035     const struct _CFStreamCallBacks *cb;
1036     if (status != kCFStreamStatusOpen && status != kCFStreamStatusReading) {
1037         return FALSE;
1038     }
1039     cb  = _CFStreamGetCallBackPtr(stream);
1040     if (cb->canRead == NULL) {
1041         return TRUE;  // No way to know without trying....
1042     } else {
1043         Boolean result;
1044         __CFBitSet(stream->flags, CALLING_CLIENT);
1045         if (cb->version < 2) {
1046             result = ((_CFStreamCBCanReadV1)(cb->canRead))((CFReadStreamRef)stream, _CFStreamGetInfoPointer(stream));
1047         } else {
1048             result = cb->canRead((CFReadStreamRef)stream, &(stream->error), _CFStreamGetInfoPointer(stream));
1049             if (stream->error) {
1050                 _CFStreamSetStatusCode(stream, kCFStreamStatusError);
1051                 _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
1052             }
1053         }
1054         __CFBitClear(stream->flags, CALLING_CLIENT);
1055         return result;
1056     }
1057 }
1058
1059 static void waitForOpen(struct _CFStream *stream);
1060 CFIndex CFReadStreamRead(CFReadStreamRef readStream, UInt8 *buffer, CFIndex bufferLength) {
1061     CF_OBJC_FUNCDISPATCHV(__kCFReadStreamTypeID, CFIndex, (NSInputStream *)readStream, read:(uint8_t *)buffer maxLength:(NSUInteger)bufferLength);
1062     struct _CFStream *stream = (struct _CFStream *)readStream;
1063     CFStreamStatus status = _CFStreamGetStatus(stream);
1064     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1065     if (status == kCFStreamStatusOpening) {
1066         __CFBitSet(stream->flags, CALLING_CLIENT);
1067         waitForOpen(stream);
1068         __CFBitClear(stream->flags, CALLING_CLIENT);
1069         status = _CFStreamGetStatus(stream);
1070     }
1071
1072     if (status != kCFStreamStatusOpen && status != kCFStreamStatusReading && status != kCFStreamStatusAtEnd) {
1073         return -1;
1074     } else  if (status == kCFStreamStatusAtEnd) {
1075         return 0;
1076     } else {
1077         Boolean atEOF;
1078         CFIndex bytesRead;
1079         __CFBitSet(stream->flags, CALLING_CLIENT);
1080         if (stream->client) {
1081             stream->client->whatToSignal &= ~kCFStreamEventHasBytesAvailable;
1082         }
1083         _CFStreamSetStatusCode(stream, kCFStreamStatusReading);
1084         if (cb->version < 2) {
1085             CFStreamError err = {0, 0};
1086             bytesRead = ((_CFStreamCBReadV1)(cb->read))((CFReadStreamRef)stream, buffer, bufferLength, &err, &atEOF, _CFStreamGetInfoPointer(stream));
1087             if (err.error != 0) _CFStreamSetStreamError(stream, &err);
1088         } else {
1089             bytesRead = cb->read((CFReadStreamRef)stream, buffer, bufferLength, &(stream->error), &atEOF, _CFStreamGetInfoPointer(stream));
1090         }
1091         if (stream->error) {
1092             bytesRead = -1;
1093             _CFStreamSetStatusCode(stream, kCFStreamStatusError);
1094             _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
1095         } else if (atEOF) {
1096             _CFStreamSetStatusCode(stream, kCFStreamStatusAtEnd);
1097             _CFStreamScheduleEvent(stream, kCFStreamEventEndEncountered);
1098         } else {
1099             _CFStreamSetStatusCode(stream, kCFStreamStatusOpen);
1100         }
1101         __CFBitClear(stream->flags, CALLING_CLIENT);
1102         return bytesRead;
1103     }
1104 }
1105
1106 CF_EXPORT const UInt8 *CFReadStreamGetBuffer(CFReadStreamRef readStream, CFIndex maxBytesToRead, CFIndex *numBytesRead) {
1107     if (CF_IS_OBJC(__kCFReadStreamTypeID, readStream)) {
1108         uint8_t *bufPtr = NULL;
1109         Boolean gotBytes = (Boolean) CF_OBJC_CALLV((NSInputStream *)readStream, getBuffer:&bufPtr length:(NSUInteger *)numBytesRead);
1110         if(gotBytes) {
1111             return (const UInt8 *)bufPtr;
1112         } else {
1113             return NULL;
1114         }
1115     }
1116     struct _CFStream *stream = (struct _CFStream *)readStream;
1117     CFStreamStatus status = _CFStreamGetStatus(stream);
1118     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1119     const UInt8 *buffer;
1120     if (status == kCFStreamStatusOpening) {
1121         __CFBitSet(stream->flags, CALLING_CLIENT);
1122         waitForOpen(stream);
1123         __CFBitClear(stream->flags, CALLING_CLIENT);
1124         status = _CFStreamGetStatus(stream);
1125     }
1126     if (status != kCFStreamStatusOpen && status != kCFStreamStatusReading && status != kCFStreamStatusAtEnd) {
1127         *numBytesRead = -1;
1128         buffer = NULL;
1129     } else  if (status == kCFStreamStatusAtEnd || cb->getBuffer == NULL) {
1130         *numBytesRead = 0;
1131         buffer = NULL;
1132     } else {
1133         Boolean atEOF;
1134         Boolean hadBytes = stream->client && (stream->client->whatToSignal & kCFStreamEventHasBytesAvailable);
1135         __CFBitSet(stream->flags, CALLING_CLIENT);
1136         if (hadBytes) {
1137             stream->client->whatToSignal &= ~kCFStreamEventHasBytesAvailable;
1138         }
1139         _CFStreamSetStatusCode(stream, kCFStreamStatusReading);
1140         if (cb->version < 2) {
1141             CFStreamError err = {0, 0};
1142             buffer = ((_CFStreamCBGetBufferV1)(cb->getBuffer))((CFReadStreamRef)stream, maxBytesToRead, numBytesRead, &err, &atEOF, _CFStreamGetInfoPointer(stream));
1143             if (err.error != 0) _CFStreamSetStreamError(stream, &err);
1144         } else {
1145             buffer = cb->getBuffer((CFReadStreamRef)stream, maxBytesToRead, numBytesRead, &(stream->error), &atEOF, _CFStreamGetInfoPointer(stream));
1146         }
1147         if (stream->error) {
1148             *numBytesRead = -1;
1149             _CFStreamSetStatusCode(stream, kCFStreamStatusError);
1150             buffer = NULL;
1151             _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
1152         } else if (atEOF) {
1153             _CFStreamSetStatusCode(stream, kCFStreamStatusAtEnd);
1154             _CFStreamScheduleEvent(stream, kCFStreamEventEndEncountered);
1155         } else {
1156             if (!buffer && hadBytes) {
1157                 stream->client->whatToSignal |= kCFStreamEventHasBytesAvailable;
1158             }
1159             _CFStreamSetStatusCode(stream, kCFStreamStatusOpen);
1160         }
1161         __CFBitClear(stream->flags, CALLING_CLIENT);
1162     }
1163     return buffer;
1164 }
1165
1166 CF_EXPORT Boolean CFWriteStreamCanAcceptBytes(CFWriteStreamRef writeStream) {
1167     CF_OBJC_FUNCDISPATCHV(__kCFWriteStreamTypeID, Boolean, (NSOutputStream *)writeStream, hasSpaceAvailable);
1168     struct _CFStream *stream = (struct _CFStream *)writeStream;
1169     CFStreamStatus status = _CFStreamGetStatus(stream);
1170     const struct _CFStreamCallBacks *cb;
1171     if (status != kCFStreamStatusOpen && status != kCFStreamStatusWriting) {
1172         return FALSE;
1173     }
1174     cb  = _CFStreamGetCallBackPtr(stream);
1175     if (cb->canWrite == NULL) {
1176         return TRUE;  // No way to know without trying....
1177     } else {
1178         Boolean result;
1179         __CFBitSet(stream->flags, CALLING_CLIENT);
1180         if (cb->version < 2) {
1181             result = ((_CFStreamCBCanWriteV1)(cb->canWrite))((CFWriteStreamRef)stream, _CFStreamGetInfoPointer(stream));
1182         } else {
1183             result = cb->canWrite((CFWriteStreamRef)stream, &(stream->error), _CFStreamGetInfoPointer(stream));
1184             if (stream->error) {
1185                 _CFStreamSetStatusCode(stream, kCFStreamStatusError);
1186                 _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
1187             }
1188         }
1189         __CFBitClear(stream->flags, CALLING_CLIENT);
1190         return result;
1191     }
1192 }
1193
1194 CF_EXPORT CFIndex CFWriteStreamWrite(CFWriteStreamRef writeStream, const UInt8 *buffer, CFIndex bufferLength) {
1195     CF_OBJC_FUNCDISPATCHV(__kCFWriteStreamTypeID, CFIndex, (NSOutputStream *)writeStream, write:(const uint8_t *)buffer maxLength:(NSUInteger)bufferLength);
1196     struct _CFStream *stream = (struct _CFStream *)writeStream;
1197     CFStreamStatus status = _CFStreamGetStatus(stream);
1198     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1199     if (status == kCFStreamStatusOpening) {
1200         __CFBitSet(stream->flags, CALLING_CLIENT);
1201         waitForOpen(stream);
1202         __CFBitClear(stream->flags, CALLING_CLIENT);
1203         status = _CFStreamGetStatus(stream);
1204     }
1205     if (status != kCFStreamStatusOpen && status != kCFStreamStatusWriting) {
1206         return -1;
1207     } else {
1208         CFIndex result;
1209         __CFBitSet(stream->flags, CALLING_CLIENT);
1210         _CFStreamSetStatusCode(stream, kCFStreamStatusWriting);
1211         if (stream->client) {
1212             stream->client->whatToSignal &= ~kCFStreamEventCanAcceptBytes;
1213         }
1214         if (cb->version < 2) {
1215             CFStreamError err = {0, 0};
1216             result = ((_CFStreamCBWriteV1)(cb->write))((CFWriteStreamRef)stream, buffer, bufferLength, &err, _CFStreamGetInfoPointer(stream));
1217             if (err.error) _CFStreamSetStreamError(stream, &err);
1218         } else {
1219             result = cb->write((CFWriteStreamRef)stream, buffer, bufferLength, &(stream->error), _CFStreamGetInfoPointer(stream));
1220         }
1221         if (stream->error) {
1222             _CFStreamSetStatusCode(stream, kCFStreamStatusError);
1223             _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
1224         } else if (result == 0) {
1225             _CFStreamSetStatusCode(stream, kCFStreamStatusAtEnd);
1226             _CFStreamScheduleEvent(stream, kCFStreamEventEndEncountered);
1227         } else {
1228             _CFStreamSetStatusCode(stream, kCFStreamStatusOpen);
1229         }
1230         __CFBitClear(stream->flags, CALLING_CLIENT);
1231         return result;
1232     }
1233 }
1234
1235 CF_PRIVATE CFTypeRef _CFStreamCopyProperty(struct _CFStream *stream, CFStringRef propertyName) {
1236     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1237     if (cb->copyProperty == NULL) {
1238         return NULL;
1239     } else {
1240         CFTypeRef result;
1241         __CFBitSet(stream->flags, CALLING_CLIENT);
1242         result = cb->copyProperty(stream, propertyName, _CFStreamGetInfoPointer(stream));
1243         __CFBitClear(stream->flags, CALLING_CLIENT);
1244         return result;
1245     }
1246 }
1247
1248 CF_EXPORT CFTypeRef CFReadStreamCopyProperty(CFReadStreamRef stream, CFStringRef propertyName) {
1249     CF_OBJC_FUNCDISPATCHV(__kCFReadStreamTypeID, CFTypeRef, (NSInputStream *)stream, propertyForKey:(NSString *)propertyName);
1250     return _CFStreamCopyProperty((struct _CFStream *)stream, propertyName);
1251 }
1252
1253 CF_EXPORT CFTypeRef CFWriteStreamCopyProperty(CFWriteStreamRef stream, CFStringRef propertyName) {
1254     CF_OBJC_FUNCDISPATCHV(__kCFWriteStreamTypeID, CFTypeRef, (NSOutputStream *)stream, propertyForKey:(NSString *)propertyName);
1255     return _CFStreamCopyProperty((struct _CFStream *)stream, propertyName);
1256 }
1257
1258 CF_PRIVATE Boolean _CFStreamSetProperty(struct _CFStream *stream, CFStringRef prop, CFTypeRef val) {
1259     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1260     if (cb->setProperty == NULL) {
1261         return FALSE;
1262     } else {
1263         Boolean result;
1264         __CFBitSet(stream->flags, CALLING_CLIENT);
1265         result = cb->setProperty(stream, prop, val, _CFStreamGetInfoPointer(stream));
1266         __CFBitClear(stream->flags, CALLING_CLIENT);
1267         return result;
1268     }
1269 }
1270
1271 CF_EXPORT
1272 Boolean CFReadStreamSetProperty(CFReadStreamRef stream, CFStringRef propertyName, CFTypeRef propertyValue) {
1273     CF_OBJC_FUNCDISPATCHV(__kCFReadStreamTypeID, Boolean, (NSInputStream *)stream, setProperty:(id)propertyValue forKey:(NSString *)propertyName);
1274     return _CFStreamSetProperty((struct _CFStream *)stream, propertyName, propertyValue);
1275 }
1276
1277 CF_EXPORT
1278 Boolean CFWriteStreamSetProperty(CFWriteStreamRef stream, CFStringRef propertyName, CFTypeRef propertyValue) {
1279     CF_OBJC_FUNCDISPATCHV(__kCFWriteStreamTypeID, Boolean, (NSOutputStream *)stream, setProperty:(id)propertyValue forKey:(NSString *)propertyName);
1280     return _CFStreamSetProperty((struct _CFStream *)stream, propertyName, propertyValue);
1281 }
1282
1283 static void _initializeClient(struct _CFStream *stream) {
1284     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1285     if (!cb->schedule) return; // Do we wish to allow this?
1286     stream->client = (struct _CFStreamClient *)CFAllocatorAllocate(CFGetAllocator(stream), sizeof(struct _CFStreamClient), 0);
1287     memset(stream->client, 0, sizeof(struct _CFStreamClient));
1288 }
1289
1290 /* If we add a setClient callback to the concrete stream callbacks, we must set/clear CALLING_CLIENT around it */
1291 CF_PRIVATE Boolean _CFStreamSetClient(struct _CFStream *stream, CFOptionFlags streamEvents, void (*clientCB)(struct _CFStream *, CFStreamEventType, void *), CFStreamClientContext *clientCallBackContext) {
1292
1293     Boolean removingClient = (streamEvents == kCFStreamEventNone || clientCB == NULL || clientCallBackContext == NULL);
1294
1295     if (removingClient) {
1296         clientCB = NULL;
1297         streamEvents = kCFStreamEventNone;
1298         clientCallBackContext = NULL;
1299     }
1300     if (!stream->client) {
1301         if (removingClient) {
1302             // We have no client now, and we've been asked to add none???
1303             return TRUE;
1304         }
1305         _initializeClient(stream);
1306         if (!stream->client) {
1307             // Asynch not supported
1308             return FALSE;
1309         }
1310     }
1311     if (stream->client->cb && stream->client->cbContext.release) {
1312         stream->client->cbContext.release(stream->client->cbContext.info);
1313     }
1314     stream->client->cb = clientCB;
1315     if (clientCallBackContext) {
1316         stream->client->cbContext.version = clientCallBackContext->version;
1317         stream->client->cbContext.retain = clientCallBackContext->retain;
1318         stream->client->cbContext.release = clientCallBackContext->release;
1319         stream->client->cbContext.copyDescription = clientCallBackContext->copyDescription;
1320         stream->client->cbContext.info = (clientCallBackContext->retain && clientCallBackContext->info) ? clientCallBackContext->retain(clientCallBackContext->info) : clientCallBackContext->info;
1321     } else {
1322         stream->client->cbContext.retain = NULL;
1323         stream->client->cbContext.release = NULL;
1324         stream->client->cbContext.copyDescription = NULL;
1325         stream->client->cbContext.info = NULL;
1326     }
1327     if (stream->client->when != streamEvents) {
1328         const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1329         stream->client->when = streamEvents;
1330         if (cb->requestEvents) {
1331             cb->requestEvents(stream, streamEvents, _CFStreamGetInfoPointer(stream));
1332         }
1333     }
1334     return TRUE;
1335 }
1336
1337 CF_EXPORT Boolean CFReadStreamSetClient(CFReadStreamRef readStream, CFOptionFlags streamEvents, CFReadStreamClientCallBack clientCB, CFStreamClientContext *clientContext) {
1338 #if defined(CFSTREAM_SUPPORTS_BRIDGING)
1339     if (CF_IS_OBJC(__kCFReadStreamTypeID, (const void *)(NSInputStream *)readStream)) {
1340         NSInputStream* is = (NSInputStream*) readStream;
1341
1342         if ([is respondsToSelector:@selector(_setCFClientFlags:callback:context:)])
1343             return [is _setCFClientFlags:streamEvents callback:clientCB context:clientContext];
1344         else {
1345             if (clientCB == NULL)
1346                 [is setDelegate:nil];
1347             else {
1348                 _CFStreamDelegate* d = [[_CFStreamDelegate alloc] initWithStreamEvents:streamEvents callback:(void*) clientCB context:clientContext];
1349                 [is setDelegate:d];
1350             }
1351             return true;
1352         }
1353     }
1354 #endif
1355
1356     streamEvents &= ~kCFStreamEventCanAcceptBytes;
1357     return _CFStreamSetClient((struct _CFStream *)readStream, streamEvents, (void (*)(struct _CFStream *, CFStreamEventType, void *))clientCB, clientContext);
1358 }
1359
1360 CF_EXPORT Boolean CFWriteStreamSetClient(CFWriteStreamRef writeStream, CFOptionFlags streamEvents, CFWriteStreamClientCallBack clientCB, CFStreamClientContext *clientContext) {
1361 #if defined(CFSTREAM_SUPPORTS_BRIDGING)
1362     if (CF_IS_OBJC(__kCFWriteStreamTypeID, (const void *)(NSInputStream *)writeStream)) {
1363         NSOutputStream* os = (NSOutputStream*) writeStream;
1364
1365         if ([os respondsToSelector:@selector(_setCFClientFlags:callback:context:)])
1366             return [os _setCFClientFlags:streamEvents callback:clientCB context:clientContext];
1367         else {
1368             if (clientCB == NULL)
1369                 [os setDelegate:nil];
1370             else {
1371                 _CFStreamDelegate* d = [[_CFStreamDelegate alloc] initWithStreamEvents:streamEvents callback:(void*) clientCB context:clientContext];
1372                 [os setDelegate:d];
1373             }
1374             return true;
1375         }
1376     }
1377 #endif
1378
1379     streamEvents &= ~kCFStreamEventHasBytesAvailable;
1380     return _CFStreamSetClient((struct _CFStream *)writeStream, streamEvents, (void (*)(struct _CFStream *, CFStreamEventType, void *))clientCB, clientContext);
1381 }
1382
1383 CF_INLINE void *_CFStreamGetClient(struct _CFStream *stream) {
1384     if (stream->client) return stream->client->cbContext.info;
1385     else return NULL;
1386 }
1387
1388 CF_EXPORT void *_CFReadStreamGetClient(CFReadStreamRef readStream) {
1389     return _CFStreamGetClient((struct _CFStream *)readStream);
1390 }
1391
1392 CF_EXPORT void *_CFWriteStreamGetClient(CFWriteStreamRef writeStream) {
1393     return _CFStreamGetClient((struct _CFStream *)writeStream);
1394 }
1395
1396
1397 CF_PRIVATE void _CFStreamScheduleWithRunLoop(struct _CFStream *stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1398     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1399
1400     if (! stream->client) {
1401         _initializeClient(stream);
1402         if (!stream->client) return; // we don't support asynch.
1403     }
1404
1405     if (! stream->client->rlSource) {
1406         /* No source, so we join the shared source group */
1407         CFTypeRef a[] = { runLoop, runLoopMode };
1408
1409         CFArrayRef runLoopAndSourceKey = CFArrayCreate(kCFAllocatorSystemDefault, a, sizeof(a) / sizeof(a[0]), &kCFTypeArrayCallBacks);
1410
1411         __CFSpinLock(&sSourceLock);
1412
1413         if (!sSharedSources)
1414             sSharedSources = CFDictionaryCreateMutable(kCFAllocatorSystemDefault, 0, &kCFTypeDictionaryKeyCallBacks, &kCFTypeDictionaryValueCallBacks);
1415
1416         CFMutableArrayRef listOfStreamsSharingASource = (CFMutableArrayRef)CFDictionaryGetValue(sSharedSources, runLoopAndSourceKey);
1417         if (listOfStreamsSharingASource) {
1418                         struct _CFStream* aStream = (struct _CFStream*) CFArrayGetValueAtIndex(listOfStreamsSharingASource, 0);
1419                         CFRunLoopSourceRef source = _CFStreamCopySource(aStream);
1420                         if (source) {
1421                                 _CFStreamSetSource(stream, source, FALSE);
1422                                 CFRelease(source);
1423                         }
1424             CFRetain(listOfStreamsSharingASource);
1425         }
1426         else {
1427             CFRunLoopSourceContext ctxt = {
1428                 0,
1429                 NULL,
1430                 CFRetain,
1431                 CFRelease,
1432                 (CFStringRef(*)(const void *))CFCopyDescription,
1433                 NULL,
1434                 NULL,
1435                 NULL,
1436                 NULL,
1437                 (void(*)(void *))_cfstream_shared_signalEventSync
1438             };
1439
1440             listOfStreamsSharingASource = CFArrayCreateMutable(kCFAllocatorSystemDefault, 0, &kCFTypeArrayCallBacks);
1441             CFDictionaryAddValue(sSharedSources, runLoopAndSourceKey, listOfStreamsSharingASource);
1442
1443             ctxt.info = listOfStreamsSharingASource;
1444
1445                         CFRunLoopSourceRef source = CFRunLoopSourceCreate(kCFAllocatorSystemDefault, 0, &ctxt);
1446                         _CFStreamSetSource(stream, source, FALSE);
1447             CFRunLoopAddSource(runLoop, source, runLoopMode);
1448                         CFRelease(source);
1449         }
1450
1451         CFArrayAppendValue(listOfStreamsSharingASource, stream);
1452         CFDictionaryAddValue(sSharedSources, stream, runLoopAndSourceKey);
1453
1454         CFRelease(runLoopAndSourceKey);
1455         CFRelease(listOfStreamsSharingASource);
1456
1457         __CFBitSet(stream->flags, SHARED_SOURCE);
1458
1459         __CFSpinUnlock(&sSourceLock);
1460     }
1461     else if (__CFBitIsSet(stream->flags, SHARED_SOURCE)) {
1462         /* We were sharing, but now we'll get our own source */
1463
1464         CFArrayRef runLoopAndSourceKey;
1465         CFMutableArrayRef listOfStreamsSharingASource;
1466         CFIndex count, i;
1467
1468         CFAllocatorRef alloc = CFGetAllocator(stream);
1469         CFRunLoopSourceContext ctxt = {
1470             0,
1471             (void *)stream,
1472             NULL,                                                                                                               // Do not use CFRetain/CFRelease callbacks here; that will cause a retain loop
1473             NULL,                                                                                                               // Do not use CFRetain/CFRelease callbacks here; that will cause a retain loop
1474             (CFStringRef(*)(const void *))CFCopyDescription,
1475             NULL,
1476             NULL,
1477             NULL,
1478             NULL,
1479             (void(*)(void *))_cfstream_solo_signalEventSync
1480         };
1481
1482         __CFSpinLock(&sSourceLock);
1483
1484         runLoopAndSourceKey = (CFArrayRef)CFRetain((CFTypeRef)CFDictionaryGetValue(sSharedSources, stream));
1485         listOfStreamsSharingASource = (CFMutableArrayRef)CFDictionaryGetValue(sSharedSources, runLoopAndSourceKey);
1486
1487         count = CFArrayGetCount(listOfStreamsSharingASource);
1488         i = CFArrayGetFirstIndexOfValue(listOfStreamsSharingASource, CFRangeMake(0, count), stream);
1489         if (i != kCFNotFound) {
1490             CFArrayRemoveValueAtIndex(listOfStreamsSharingASource, i);
1491             count--;
1492         }
1493
1494         if (count == 0) {
1495                         CFRunLoopSourceRef source = _CFStreamCopySource(stream);
1496                         if (source) {
1497                                 CFRunLoopRemoveSource((CFRunLoopRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 0), source, (CFStringRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 1));
1498                                 CFRelease(source);
1499                         }
1500             CFDictionaryRemoveValue(sSharedSources, runLoopAndSourceKey);
1501         }
1502
1503         CFDictionaryRemoveValue(sSharedSources, stream);
1504
1505                 _CFStreamSetSource(stream, NULL, count == 0);
1506
1507         __CFBitClear(stream->flags, SHARED_SOURCE);
1508
1509         __CFSpinUnlock(&sSourceLock);
1510
1511                 CFRunLoopSourceRef source = CFRunLoopSourceCreate(alloc, 0, &ctxt);
1512                 _CFStreamSetSource(stream, source, FALSE);
1513         CFRunLoopAddSource((CFRunLoopRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 0), source, (CFStringRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 1));
1514         CFRelease(runLoopAndSourceKey);
1515
1516         CFRunLoopAddSource(runLoop, source, runLoopMode);
1517
1518                 CFRelease(source);
1519     } else {
1520         /* We're not sharing, so just add the source to the rl & mode */
1521                 CFRunLoopSourceRef source = _CFStreamCopySource(stream);
1522                 if (source) {
1523                         CFRunLoopAddSource(runLoop, source, runLoopMode);
1524                         CFRelease(source);
1525                 }
1526     }
1527
1528     _CFStreamLock(stream);
1529     if (!stream->client->runLoopsAndModes) {
1530         stream->client->runLoopsAndModes = CFArrayCreateMutable(CFGetAllocator(stream), 0, &kCFTypeArrayCallBacks);
1531     }
1532     CFArrayAppendValue(stream->client->runLoopsAndModes, runLoop);
1533     CFArrayAppendValue(stream->client->runLoopsAndModes, runLoopMode);
1534     checkRLMArray(stream->client->runLoopsAndModes);
1535     _CFStreamUnlock(stream);
1536
1537     if (cb->schedule) {
1538         __CFBitSet(stream->flags, CALLING_CLIENT);
1539         cb->schedule(stream, runLoop, runLoopMode, _CFStreamGetInfoPointer(stream));
1540         __CFBitClear(stream->flags, CALLING_CLIENT);
1541     }
1542
1543     /*
1544      * If we've got events pending, we need to wake up and signal
1545      */
1546     if (stream->client && stream->client->whatToSignal != 0) {
1547                 CFRunLoopSourceRef source = _CFStreamCopySource(stream);
1548                 if (source) {
1549                         CFRunLoopSourceSignal(source);
1550                         CFRelease(source);
1551             _wakeUpRunLoop(stream);
1552         }
1553     }
1554 }
1555
1556 CF_EXPORT void CFReadStreamScheduleWithRunLoop(CFReadStreamRef stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1557 #if defined(CFSTREAM_SUPPORTS_BRIDGING)
1558     if (CF_IS_OBJC(__kCFReadStreamTypeID, (const void *)(NSInputStream *)stream)) {
1559         NSInputStream* is  = (NSInputStream*) stream;
1560         if ([is respondsToSelector:@selector(_scheduleInCFRunLoop:forMode:)])
1561             [is _scheduleInCFRunLoop:runLoop forMode:runLoopMode];
1562         else
1563             [is scheduleInRunLoop:cfToNSRL(runLoop) forMode:(id) runLoopMode];
1564         return;
1565     }
1566 #endif
1567
1568     _CFStreamScheduleWithRunLoop((struct _CFStream *)stream, runLoop, runLoopMode);
1569 }
1570
1571 CF_EXPORT void CFWriteStreamScheduleWithRunLoop(CFWriteStreamRef stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1572 #if defined(CFSTREAM_SUPPORTS_BRIDGING)
1573     if (CF_IS_OBJC(__kCFWriteStreamTypeID, (const void *)(NSOutputStream *)stream)) {
1574         NSOutputStream* os  = (NSOutputStream*) stream;
1575         if ([os respondsToSelector:@selector(_scheduleInCFRunLoop:forMode:)])
1576             [os _scheduleInCFRunLoop:runLoop forMode:runLoopMode];
1577         else
1578             [os scheduleInRunLoop:cfToNSRL(runLoop) forMode:(id) runLoopMode];
1579         return;
1580     }
1581 #endif
1582
1583     _CFStreamScheduleWithRunLoop((struct _CFStream *)stream, runLoop, runLoopMode);
1584 }
1585
1586
1587 CF_PRIVATE void _CFStreamUnscheduleFromRunLoop(struct _CFStream *stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1588     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1589     if (!stream->client) return;
1590     if (!stream->client->rlSource) return;
1591
1592     if (!__CFBitIsSet(stream->flags, SHARED_SOURCE)) {
1593                 CFRunLoopSourceRef source = _CFStreamCopySource(stream);
1594                 if (source) {
1595                         CFRunLoopRemoveSource(runLoop, source, runLoopMode);
1596                         CFRelease(source);
1597                 }
1598     } else {
1599         CFArrayRef runLoopAndSourceKey;
1600         CFMutableArrayRef list;
1601         CFIndex count, i;
1602
1603         __CFSpinLock(&sSourceLock);
1604
1605         runLoopAndSourceKey = (CFArrayRef)CFDictionaryGetValue(sSharedSources, stream);
1606         list = (CFMutableArrayRef)CFDictionaryGetValue(sSharedSources, runLoopAndSourceKey);
1607
1608         count = CFArrayGetCount(list);
1609         i = CFArrayGetFirstIndexOfValue(list, CFRangeMake(0, count), stream);
1610         if (i != kCFNotFound) {
1611             CFArrayRemoveValueAtIndex(list, i);
1612             count--;
1613         }
1614
1615         if (count == 0) {
1616                         CFRunLoopSourceRef source = _CFStreamCopySource(stream);
1617                         if (source) {
1618                                 CFRunLoopRemoveSource(runLoop, source, runLoopMode);
1619                                 CFRelease(source);
1620                         }
1621             CFDictionaryRemoveValue(sSharedSources, runLoopAndSourceKey);
1622         }
1623
1624         CFDictionaryRemoveValue(sSharedSources, stream);
1625
1626         _CFStreamSetSource(stream, NULL, count == 0);
1627
1628         __CFBitClear(stream->flags, SHARED_SOURCE);
1629
1630         __CFSpinUnlock(&sSourceLock);
1631     }
1632
1633     _CFStreamLock(stream);
1634     _CFStreamRemoveRunLoopAndModeFromArray(stream->client->runLoopsAndModes, runLoop, runLoopMode);
1635     checkRLMArray(stream->client->runLoopsAndModes);
1636     _CFStreamUnlock(stream);
1637
1638     if (cb->unschedule) {
1639         cb->unschedule(stream, runLoop, runLoopMode, _CFStreamGetInfoPointer(stream));
1640     }
1641 }
1642
1643 CF_EXPORT void CFReadStreamUnscheduleFromRunLoop(CFReadStreamRef stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1644 #if defined(CFSTREAM_SUPPORTS_BRIDGING)
1645     if (CF_IS_OBJC(__kCFReadStreamTypeID, (const void *)(NSInputStream *)stream)) {
1646         NSInputStream* is  = (NSInputStream*) stream;
1647         if ([is respondsToSelector:@selector(_unscheduleFromCFRunLoop:forMode:)])
1648             [is _unscheduleFromCFRunLoop:runLoop forMode:runLoopMode];
1649         else
1650             [is removeFromRunLoop:cfToNSRL(runLoop) forMode:(id) runLoopMode];
1651         return;
1652     }
1653 #endif
1654
1655     _CFStreamUnscheduleFromRunLoop((struct _CFStream *)stream, runLoop, runLoopMode);
1656 }
1657
1658 void CFWriteStreamUnscheduleFromRunLoop(CFWriteStreamRef stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1659 #if defined(CFSTREAM_SUPPORTS_BRIDGING)
1660     if (CF_IS_OBJC(__kCFWriteStreamTypeID, (const void *)(NSOutputStream *)stream)) {
1661         NSOutputStream* os  = (NSOutputStream*) stream;
1662         if ([os respondsToSelector:@selector(_unscheduleFromCFRunLoop:forMode:)])
1663             [os _unscheduleFromCFRunLoop:runLoop forMode:runLoopMode];
1664         else
1665             [os removeFromRunLoop:cfToNSRL(runLoop) forMode:(id) runLoopMode];
1666         return;
1667     }
1668 #endif
1669
1670     _CFStreamUnscheduleFromRunLoop((struct _CFStream *)stream, runLoop, runLoopMode);
1671 }
1672
1673 static CFRunLoopRef sLegacyRL = NULL;
1674
1675 static void _perform(void* info)
1676 {
1677 }
1678
1679 static void* _legacyStreamRunLoop_workThread(void* arg)
1680 {
1681     sLegacyRL = CFRunLoopGetCurrent();
1682
1683 #if defined(LOG_STREAM)
1684     fprintf(stderr, "Creating Schedulingset emulation thread.  Runloop: %p\n", sLegacyRL);
1685 #endif
1686
1687     CFStringRef s = CFStringCreateWithFormat(kCFAllocatorDefault, NULL, CFSTR("<< CFStreamLegacySource for Runloop %p >>"), sLegacyRL);
1688
1689     CFRunLoopSourceContext ctxt = {
1690         0,
1691         (void*) s,
1692         CFRetain,
1693         CFRelease,
1694         CFCopyDescription,
1695         CFEqual,
1696         CFHash,
1697         NULL,
1698         NULL,
1699         _perform
1700     };
1701
1702     CFRunLoopSourceRef rls = CFRunLoopSourceCreate(kCFAllocatorDefault, 0, &ctxt);
1703     CFRelease(s);
1704
1705     CFRunLoopAddSource(sLegacyRL, rls, kCFRunLoopDefaultMode);
1706     CFRelease(rls);
1707
1708     dispatch_semaphore_signal(*(dispatch_semaphore_t*) arg);
1709     arg = NULL;
1710
1711     while (true) {
1712         SInt32 why = CFRunLoopRunInMode(kCFRunLoopDefaultMode, 1E30, true);
1713
1714         (void) why;
1715 #if defined(LOG_STREAM)
1716         switch (why) {
1717             case kCFRunLoopRunFinished:
1718                 fprintf(stderr, "WOKE: kCFRunLoopRunFinished\n");
1719                 break;
1720             case kCFRunLoopRunStopped:
1721                 fprintf(stderr, "WOKE: kCFRunLoopRunStopped\n");
1722                 break;
1723             case kCFRunLoopRunTimedOut:
1724                 fprintf(stderr, "WOKE: kCFRunLoopRunTimedOut\n");
1725                 break;
1726             case kCFRunLoopRunHandledSource:
1727                 fprintf(stderr, "WOKE: kCFRunLoopRunHandledSource\n");
1728                 break;
1729         }
1730 #endif
1731     }
1732
1733     return NULL;
1734 }
1735 static CFRunLoopRef _legacyStreamRunLoop()
1736 {
1737     static dispatch_once_t sOnce = 0;
1738
1739     dispatch_once(&sOnce, ^{
1740
1741         if (sLegacyRL == NULL) {
1742             dispatch_semaphore_t sem = dispatch_semaphore_create(0);
1743
1744             pthread_attr_t attr;
1745             pthread_attr_init(&attr);
1746             pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
1747             pthread_t workThread;
1748             (void) pthread_create(&workThread, &attr, _legacyStreamRunLoop_workThread, &sem);
1749             pthread_attr_destroy(&attr);
1750
1751             dispatch_semaphore_wait(sem, DISPATCH_TIME_FOREVER);
1752             dispatch_release(sem);
1753         }
1754     });
1755
1756     return sLegacyRL;
1757 }
1758
1759 static dispatch_queue_t _CFStreamCopyDispatchQueue(struct _CFStream* stream)
1760 {
1761     dispatch_queue_t result = NULL;
1762
1763     _CFStreamLock(stream);
1764     if (stream->client) {
1765         result = stream->queue;
1766         if (result)
1767             dispatch_retain(result);
1768     }
1769     _CFStreamUnlock(stream);
1770
1771     return result;
1772 }
1773
1774 static void _CFStreamSetDispatchQueue(struct _CFStream* stream, dispatch_queue_t q)
1775 {
1776     CFArrayRef rlm = _CFStreamCopyRunLoopsAndModes(stream);
1777     if (rlm) {
1778         CFIndex count = CFArrayGetCount(rlm);
1779         for (CFIndex i = 0;  i < count;  i += 2) {
1780             CFRunLoopRef rl = (CFRunLoopRef) CFArrayGetValueAtIndex(rlm, i);
1781             CFStringRef mode = (CFStringRef) CFArrayGetValueAtIndex(rlm, i + 1);
1782             _CFStreamUnscheduleFromRunLoop(stream, rl, mode);
1783         }
1784         CFRelease(rlm);
1785     }
1786
1787     if (q == NULL) {
1788         _CFStreamLock(stream);
1789         if (stream->client) {
1790             if (stream->queue)
1791                 dispatch_release(stream->queue);
1792             stream->queue = NULL;
1793         }
1794         _CFStreamUnlock(stream);
1795     } else {
1796         _CFStreamScheduleWithRunLoop(stream, _legacyStreamRunLoop(), kCFRunLoopDefaultMode);
1797
1798         _CFStreamLock(stream);
1799         if (stream->client) {
1800             if (stream->queue != q) {
1801                 if (stream->queue)
1802                     dispatch_release(stream->queue);
1803                 stream->queue = q;
1804                 if (stream->queue)
1805                     dispatch_retain(stream->queue);
1806             }
1807         }
1808
1809         _CFStreamUnlock(stream);
1810     }
1811 }
1812
1813 void CFReadStreamSetDispatchQueue(CFReadStreamRef stream, dispatch_queue_t q)
1814 {
1815     _CFStreamSetDispatchQueue((struct _CFStream*) stream, q);
1816 }
1817
1818 void CFWriteStreamSetDispatchQueue(CFWriteStreamRef stream, dispatch_queue_t q)
1819 {
1820     _CFStreamSetDispatchQueue((struct _CFStream*) stream, q);
1821 }
1822
1823 dispatch_queue_t CFReadStreamCopyDispatchQueue(CFReadStreamRef stream)
1824 {
1825     return _CFStreamCopyDispatchQueue((struct _CFStream*) stream);
1826 }
1827
1828 dispatch_queue_t CFWriteStreamCopyDispatchQueue(CFWriteStreamRef stream)
1829 {
1830     return _CFStreamCopyDispatchQueue((struct _CFStream*) stream);
1831 }
1832
1833
1834 static void waitForOpen(struct _CFStream *stream) {
1835     CFRunLoopRef runLoop = CFRunLoopGetCurrent();
1836     CFStringRef privateMode = CFSTR("_kCFStreamBlockingOpenMode");
1837     _CFStreamScheduleWithRunLoop(stream, runLoop, privateMode);
1838     // We cannot call _CFStreamGetStatus, because that tries to set/clear CALLING_CLIENT, which should be set around this entire call (we're within a call from the client).  This should be o.k., because we're running the run loop, so our status code should be being updated in a timely fashion....
1839     while (__CFStreamGetStatus(stream) == kCFStreamStatusOpening) {
1840         CFRunLoopRunInMode(privateMode, 1e+20, TRUE);
1841     }
1842     _CFStreamUnscheduleFromRunLoop(stream, runLoop, privateMode);
1843 }
1844
1845 CF_EXPORT CFArrayRef _CFReadStreamGetRunLoopsAndModes(CFReadStreamRef readStream) {
1846     return _CFStreamGetRunLoopsAndModes((struct _CFStream *)readStream);
1847 }
1848
1849 CF_EXPORT CFArrayRef _CFWriteStreamGetRunLoopsAndModes(CFWriteStreamRef writeStream) {
1850     return _CFStreamGetRunLoopsAndModes((struct _CFStream *)writeStream);
1851 }
1852
1853 CF_EXPORT CFArrayRef _CFReadStreamCopyRunLoopsAndModes(CFReadStreamRef readStream) {
1854     return _CFStreamCopyRunLoopsAndModes((struct _CFStream *)readStream);
1855 }
1856
1857 CF_EXPORT CFArrayRef _CFWriteStreamCopyRunLoopsAndModes(CFWriteStreamRef writeStream) {
1858     return _CFStreamCopyRunLoopsAndModes((struct _CFStream *)writeStream);
1859 }
1860
1861 CF_EXPORT void CFReadStreamSignalEvent(CFReadStreamRef stream, CFStreamEventType event, const void *error) {
1862     _CFStreamSignalEvent((struct _CFStream *)stream, event, (CFErrorRef)error, TRUE);
1863 }
1864
1865 CF_EXPORT void CFWriteStreamSignalEvent(CFWriteStreamRef stream, CFStreamEventType event, const void *error) {
1866     _CFStreamSignalEvent((struct _CFStream *)stream, event, (CFErrorRef)error, TRUE);
1867 }
1868
1869 CF_EXPORT void _CFReadStreamSignalEventDelayed(CFReadStreamRef stream, CFStreamEventType event, const void *error) {
1870     _CFStreamSignalEvent((struct _CFStream *)stream, event, (CFErrorRef)error, FALSE);
1871 }
1872
1873 CF_EXPORT void _CFReadStreamClearEvent(CFReadStreamRef readStream, CFStreamEventType event) {
1874     struct _CFStream *stream = (struct _CFStream *)readStream;
1875     if (stream->client) {
1876         stream->client->whatToSignal &= ~event;
1877     }
1878 }
1879
1880 CF_EXPORT void _CFWriteStreamSignalEventDelayed(CFWriteStreamRef stream, CFStreamEventType event, const void *error) {
1881     _CFStreamSignalEvent((struct _CFStream *)stream, event, (CFErrorRef)error, FALSE);
1882 }
1883
1884 CF_EXPORT void *CFReadStreamGetInfoPointer(CFReadStreamRef stream) {
1885     return _CFStreamGetInfoPointer((struct _CFStream *)stream);
1886 }
1887
1888 CF_EXPORT void *CFWriteStreamGetInfoPointer(CFWriteStreamRef stream) {
1889     return _CFStreamGetInfoPointer((struct _CFStream *)stream);
1890 }
1891
1892 /* CF_EXPORT */
1893 void _CFStreamSourceScheduleWithRunLoop(CFRunLoopSourceRef source, CFMutableArrayRef runLoopsAndModes, CFRunLoopRef runLoop, CFStringRef runLoopMode)
1894 {
1895     CFIndex count;
1896     CFRange range;
1897
1898     checkRLMArray(runLoopsAndModes);
1899
1900     count = CFArrayGetCount(runLoopsAndModes);
1901     range = CFRangeMake(0, count);
1902
1903     while (range.length) {
1904
1905         CFIndex i = CFArrayGetFirstIndexOfValue(runLoopsAndModes, range, runLoop);
1906
1907         if (i == kCFNotFound)
1908             break;
1909
1910         if (CFEqual(CFArrayGetValueAtIndex(runLoopsAndModes, i + 1), runLoopMode))
1911             return;
1912
1913         range.location = i + 2;
1914         range.length = count - range.location;
1915     }
1916
1917         // Add the new values.
1918     CFArrayAppendValue(runLoopsAndModes, runLoop);
1919     CFArrayAppendValue(runLoopsAndModes, runLoopMode);
1920
1921         // Schedule the source on the new loop and mode.
1922     if (source)
1923         CFRunLoopAddSource(runLoop, source, runLoopMode);
1924 }
1925
1926
1927 /* CF_EXPORT */
1928 void _CFStreamSourceUnscheduleFromRunLoop(CFRunLoopSourceRef source, CFMutableArrayRef runLoopsAndModes, CFRunLoopRef runLoop, CFStringRef runLoopMode)
1929 {
1930     CFIndex count;
1931         CFRange range;
1932
1933     count = CFArrayGetCount(runLoopsAndModes);
1934     range = CFRangeMake(0, count);
1935
1936     checkRLMArray(runLoopsAndModes);
1937
1938         while (range.length) {
1939
1940                 CFIndex i = CFArrayGetFirstIndexOfValue(runLoopsAndModes, range, runLoop);
1941
1942                 // If not found, it's not scheduled on it.
1943                 if (i == kCFNotFound)
1944                         return;
1945
1946                 // Make sure it is scheduled in this mode.
1947                 if (CFEqual(CFArrayGetValueAtIndex(runLoopsAndModes, i + 1), runLoopMode)) {
1948
1949                         // Remove mode and runloop from the list.
1950             CFArrayReplaceValues(runLoopsAndModes, CFRangeMake(i, 2), NULL, 0);
1951
1952             // Remove it from the runloop.
1953             if (source)
1954                 CFRunLoopRemoveSource(runLoop, source, runLoopMode);
1955
1956                         return;
1957                 }
1958
1959         range.location = i + 2;
1960         range.length = count - range.location;
1961         }
1962 }
1963
1964
1965 /* CF_EXPORT */
1966 void _CFStreamSourceScheduleWithAllRunLoops(CFRunLoopSourceRef source, CFArrayRef runLoopsAndModes)
1967 {
1968     CFIndex i, count = CFArrayGetCount(runLoopsAndModes);
1969
1970     if (!source)
1971         return;
1972
1973     checkRLMArray(runLoopsAndModes);
1974
1975     for (i = 0; i < count; i += 2) {
1976
1977         // Make sure it's scheduled on all the right loops and modes.
1978         // Go through the array adding the source to all loops and modes.
1979         CFRunLoopAddSource((CFRunLoopRef)CFArrayGetValueAtIndex(runLoopsAndModes, i),
1980                             source,
1981                             (CFStringRef)CFArrayGetValueAtIndex(runLoopsAndModes, i + 1));
1982         }
1983 }
1984
1985
1986 /* CF_EXPORT */
1987 void _CFStreamSourceUncheduleFromAllRunLoops(CFRunLoopSourceRef source, CFArrayRef runLoopsAndModes)
1988 {
1989     CFIndex i, count = CFArrayGetCount(runLoopsAndModes);
1990
1991     if (!source)
1992         return;
1993
1994     checkRLMArray(runLoopsAndModes);
1995
1996     for (i = 0; i < count; i += 2) {
1997
1998         // Go through the array removing the source from all loops and modes.
1999         CFRunLoopRemoveSource((CFRunLoopRef)CFArrayGetValueAtIndex(runLoopsAndModes, i),
2000                               source,
2001                               (CFStringRef)CFArrayGetValueAtIndex(runLoopsAndModes, i + 1));
2002         }
2003 }
2004
2005 Boolean _CFStreamRemoveRunLoopAndModeFromArray(CFMutableArrayRef runLoopsAndModes, CFRunLoopRef rl, CFStringRef mode) {
2006     CFIndex idx, cnt;
2007     Boolean found = FALSE;
2008
2009     if (!runLoopsAndModes) return FALSE;
2010
2011     checkRLMArray(runLoopsAndModes);
2012
2013     cnt = CFArrayGetCount(runLoopsAndModes);
2014     for (idx = 0; idx + 1 < cnt; idx += 2) {
2015         if (CFEqual(CFArrayGetValueAtIndex(runLoopsAndModes, idx), rl) && CFEqual(CFArrayGetValueAtIndex(runLoopsAndModes, idx + 1), mode)) {
2016             CFArrayRemoveValueAtIndex(runLoopsAndModes, idx);
2017             CFArrayRemoveValueAtIndex(runLoopsAndModes, idx);
2018             found = TRUE;
2019             break;
2020         }
2021     }
2022     return found;
2023 }
2024
2025 // Used by NSStream to properly allocate the bridged objects
2026 CF_EXPORT CFIndex _CFStreamInstanceSize(void) {
2027     return sizeof(struct _CFStream);
2028 }
2029
2030 #if DEPLOYMENT_TARGET_WINDOWS
2031 void __CFStreamCleanup(void) {
2032     __CFSpinLock(&sSourceLock);
2033     if (sSharedSources) {
2034         CFIndex count = CFDictionaryGetCount(sSharedSources);
2035         if (count == 0) {
2036             // Only release if empty.  If it's still holding streams (which would be a client
2037             // bug leak), freeing this dict would free the streams, which then need to access the
2038             // dict to remove themselves, which leads to a deadlock.
2039             CFRelease(sSharedSources);
2040             sSharedSources = NULL;
2041         } else {
2042             const void ** keys = (const void **)malloc(sizeof(const void *) * count);
2043 #if defined(DEBUG)
2044             int i;
2045 #endif
2046             CFDictionaryGetKeysAndValues(sSharedSources, keys, NULL);
2047              fprintf(stderr, "*** CFNetwork is shutting down, but %ld streams are still scheduled.\n", count);
2048 #if defined(DEBUG)
2049             for (i = 0; i < count;i ++) {
2050                 if ((CFGetTypeID(keys[i]) == __kCFReadStreamTypeID) || (CFGetTypeID(keys[i]) == __kCFWriteStreamTypeID)) {
2051                     CFShow(keys[i]);
2052                 }
2053             }
2054 #endif
2055         }
2056     }
2057     __CFSpinUnlock(&sSourceLock);
2058 }
2059 #endif
2060