CFStream.c

   1 /*
   2  * Copyright (c) 2012 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-2011, 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 CFStreamAux {
  41         CFSpinLock_t streamLock;
  42     CFArrayRef previousRunloopsAndModes;
  43 };
  44
  45 enum {
  46         MIN_STATUS_CODE_BIT     = 0,
  47         // ..status bits...
  48         MAX_STATUS_CODE_BIT     = 4,
  49
  50         CONSTANT_CALLBACKS      = 5,
  51         CALLING_CLIENT          = 6,    // MUST remain 6 since it's value is used elsewhere.
  52
  53     HAVE_CLOSED                 = 7,
  54
  55     // Values above used to be defined and others may rely on their values
  56
  57     // Values below should not matter if they are re-ordered or shift
  58
  59     SHARED_SOURCE
  60 };
  61
  62
  63 /* 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
  64     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  */
  65 // Used in CFNetwork too
  66
  67 /* sSharesSources holds two mappings, one the inverse of the other, between a stream and the
  68    RunLoop+RunLoopMode pair that it's scheduled in.  If the stream is scheduled in more than
  69    one loop or mode, it can't share RunLoopSources with others, and is not in this dict.
  70 */
  71 static CFSpinLock_t sSourceLock = CFSpinLockInit;
  72 static CFMutableDictionaryRef sSharedSources = NULL;
  73
  74 static CFTypeID __kCFReadStreamTypeID = _kCFRuntimeNotATypeID;
  75 static CFTypeID __kCFWriteStreamTypeID = _kCFRuntimeNotATypeID;
  76
  77 // Just reads the bits, for those cases where we don't want to go through any callback checking
  78 #define __CFStreamGetStatus(x) __CFBitfieldGetValue((x)->flags, MAX_STATUS_CODE_BIT, MIN_STATUS_CODE_BIT)
  79
  80 __private_extern__ CFStreamStatus _CFStreamGetStatus(struct _CFStream *stream);
  81 static Boolean _CFStreamRemoveRunLoopAndModeFromArray(CFMutableArrayRef runLoopsAndModes, CFRunLoopRef rl, CFStringRef mode);
  82 static void _wakeUpRunLoop(struct _CFStream *stream);
  83
  84 CF_INLINE void checkRLMArray(CFArrayRef arr)
  85 {
  86 #if defined(DEBUG)
  87     assert(arr == NULL || (CFArrayGetCount(arr) % 2) == 0);
  88 #endif
  89 }
  90
  91 CF_INLINE void* _CFStreamCreateReserved(CFAllocatorRef alloc) {
  92         struct CFStreamAux* aux = (struct CFStreamAux*) CFAllocatorAllocate(alloc, sizeof(struct CFStreamAux), 0);
  93         if (aux) {
  94                 aux->streamLock = CFSpinLockInit;
  95         aux->previousRunloopsAndModes = NULL;
  96         }
  97         return aux;
  98 }
  99
 100 CF_INLINE void _CFStreamDestroyReserved(CFAllocatorRef alloc, void* aux) {
 101     struct CFStreamAux* paux = (struct CFStreamAux*) aux;
 102     if (paux->previousRunloopsAndModes)
 103         CFRelease(paux->previousRunloopsAndModes);
 104         CFAllocatorDeallocate(alloc, aux);
 105 }
 106
 107 CF_INLINE struct CFStreamAux* _CFStreamGetAuxRecord(struct _CFStream* stream) {
 108         return (struct CFStreamAux*) stream->_reserved1;
 109 }
 110
 111 CF_INLINE void _CFStreamLock(struct _CFStream* stream) {
 112         __CFSpinLock(&_CFStreamGetAuxRecord(stream)->streamLock);
 113 }
 114
 115 CF_INLINE void _CFStreamUnlock(struct _CFStream* stream) {
 116         __CFSpinUnlock(&_CFStreamGetAuxRecord(stream)->streamLock);
 117 }
 118
 119 CF_INLINE CFRunLoopSourceRef _CFStreamCopySource(struct _CFStream* stream) {
 120         CFRunLoopSourceRef source = NULL;
 121
 122         if (stream) {
 123                 _CFStreamLock(stream);
 124
 125                 if (stream->client)
 126                         source = stream->client->rlSource;
 127
 128                 if (source)
 129                         CFRetain(source);
 130
 131                 _CFStreamUnlock(stream);
 132         }
 133
 134         return source;
 135 }
 136
 137 CF_INLINE void _CFStreamSetSource(struct _CFStream* stream, CFRunLoopSourceRef source, Boolean invalidateOldSource) {
 138         CFRunLoopSourceRef oldSource = NULL;
 139
 140         if (stream) {
 141                 _CFStreamLock(stream);
 142                 if (stream->client) {
 143                         oldSource = stream->client->rlSource;
 144                         if (oldSource != NULL)
 145                                 CFRetain(oldSource);
 146
 147                         stream->client->rlSource = source;
 148                         if (source != NULL)
 149                                 CFRetain(source);
 150                 }
 151                 _CFStreamUnlock(stream);
 152         }
 153
 154         if (oldSource) {
 155                 // Lose our extra retain
 156                 CFRelease(oldSource);
 157
 158                 if (invalidateOldSource)
 159                         CFRunLoopSourceInvalidate(oldSource);
 160
 161                 // And lose the one that held it in our stream as well
 162                 CFRelease(oldSource);
 163         }
 164 }
 165
 166 CF_INLINE const struct _CFStreamCallBacks *_CFStreamGetCallBackPtr(struct _CFStream *stream) {
 167     return stream->callBacks;
 168 }
 169
 170 CF_INLINE void _CFStreamSetStatusCode(struct _CFStream *stream, CFStreamStatus newStatus) {
 171     CFStreamStatus status = __CFStreamGetStatus(stream);
 172     if (((status != kCFStreamStatusClosed) && (status != kCFStreamStatusError)) ||
 173         ((status == kCFStreamStatusClosed) && (newStatus == kCFStreamStatusError)))
 174     {
 175         __CFBitfieldSetValue(stream->flags, MAX_STATUS_CODE_BIT, MIN_STATUS_CODE_BIT, newStatus);
 176     }
 177 }
 178
 179 CF_INLINE void _CFStreamScheduleEvent(struct _CFStream *stream, CFStreamEventType event) {
 180     if (stream->client && (stream->client->when & event)) {
 181                 CFRunLoopSourceRef source = _CFStreamCopySource(stream);
 182         if (source) {
 183             stream->client->whatToSignal |= event;
 184
 185                         CFRunLoopSourceSignal(source);
 186                         CFRelease(source);
 187                 _wakeUpRunLoop(stream);
 188             }
 189         }
 190 }
 191
 192 CF_INLINE void _CFStreamSetStreamError(struct _CFStream *stream, CFStreamError *err) {
 193     if (!stream->error) {
 194         stream->error = (CFErrorRef)CFAllocatorAllocate(CFGetAllocator(stream), sizeof(CFStreamError), 0);
 195     }
 196     memmove(stream->error, err, sizeof(CFStreamError));
 197 }
 198
 199 static CFStringRef __CFStreamCopyDescription(CFTypeRef cf) {
 200     struct _CFStream *stream = (struct _CFStream *)cf;
 201     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
 202     CFStringRef contextDescription;
 203     CFStringRef desc;
 204     if (cb->copyDescription) {
 205         if (cb->version == 0) {
 206             contextDescription = ((CFStringRef(*)(void *))cb->copyDescription)(_CFStreamGetInfoPointer(stream));
 207         } else {
 208             contextDescription = cb->copyDescription(stream, _CFStreamGetInfoPointer(stream));
 209         }
 210     } else {
 211         contextDescription = CFStringCreateWithFormat(CFGetAllocator(stream), NULL, CFSTR("info = %p"), _CFStreamGetInfoPointer(stream));
 212     }
 213     if (CFGetTypeID(cf) == __kCFReadStreamTypeID) {
 214         desc = CFStringCreateWithFormat(CFGetAllocator(stream), NULL, CFSTR("<CFReadStream %p>{%@}"), stream, contextDescription);
 215     } else {
 216         desc = CFStringCreateWithFormat(CFGetAllocator(stream), NULL, CFSTR("<CFWriteStream %p>{%@}"), stream, contextDescription);
 217     }
 218     CFRelease(contextDescription);
 219     return desc;
 220 }
 221
 222 static void _CFStreamDetachSource(struct _CFStream* stream) {
 223         if (stream && stream->client && stream->client->rlSource) {
 224         if (!__CFBitIsSet(stream->flags, SHARED_SOURCE)) {
 225                         _CFStreamSetSource(stream, NULL, TRUE);
 226         }
 227         else {
 228
 229             CFArrayRef runLoopAndSourceKey;
 230             CFMutableArrayRef list;
 231                         CFIndex count;
 232                         CFIndex i;
 233
 234             __CFSpinLock(&sSourceLock);
 235
 236             runLoopAndSourceKey = (CFArrayRef)CFDictionaryGetValue(sSharedSources, stream);
 237             list = (CFMutableArrayRef)CFDictionaryGetValue(sSharedSources, runLoopAndSourceKey);
 238
 239                         count = CFArrayGetCount(list);
 240                         i = CFArrayGetFirstIndexOfValue(list, CFRangeMake(0, count), stream);
 241             if (i != kCFNotFound) {
 242                 CFArrayRemoveValueAtIndex(list, i);
 243                                 count--;
 244             }
 245
 246             CFAssert(CFArrayGetFirstIndexOfValue(list, CFRangeMake(0, CFArrayGetCount(list)), stream) == kCFNotFound, __kCFLogAssertion, "CFStreamClose: stream found twice in its shared source's list");
 247
 248                         if (count == 0) {
 249                                 CFRunLoopSourceRef source = _CFStreamCopySource(stream);
 250                                 if (source) {
 251                                         CFRunLoopRemoveSource((CFRunLoopRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 0), source, (CFStringRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 1));
 252                                         CFRelease(source);
 253                                 }
 254                 CFDictionaryRemoveValue(sSharedSources, runLoopAndSourceKey);
 255             }
 256
 257             CFDictionaryRemoveValue(sSharedSources, stream);
 258
 259                         _CFStreamSetSource(stream, NULL, count == 0);
 260
 261             __CFBitClear(stream->flags, SHARED_SOURCE);
 262
 263             __CFSpinUnlock(&sSourceLock);
 264         }
 265     }
 266 }
 267
 268 __private_extern__ void _CFStreamClose(struct _CFStream *stream) {
 269     CFStreamStatus status = _CFStreamGetStatus(stream);
 270     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
 271     if (status == kCFStreamStatusNotOpen || status == kCFStreamStatusClosed || (status == kCFStreamStatusError && __CFBitIsSet(stream->flags, HAVE_CLOSED))) {
 272         // Stream is not open from the client's perspective; do not callout and do not update our status to "closed"
 273         return;
 274     }
 275     __CFBitSet(stream->flags, HAVE_CLOSED);
 276     __CFBitSet(stream->flags, CALLING_CLIENT);
 277     if (cb->close) {
 278         cb->close(stream, _CFStreamGetInfoPointer(stream));
 279     }
 280     if (stream->client) {
 281         _CFStreamDetachSource(stream);
 282     }
 283     _CFStreamSetStatusCode(stream, kCFStreamStatusClosed);
 284     __CFBitClear(stream->flags, CALLING_CLIENT);
 285 }
 286
 287 //static int numStreamInstances = 0;
 288
 289 static void __CFStreamDeallocate(CFTypeRef cf) {
 290     struct _CFStream *stream = (struct _CFStream *)cf;
 291     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
 292     CFAllocatorRef alloc = CFGetAllocator(stream);
 293 //    numStreamInstances --;
 294
 295     // Close the stream
 296     _CFStreamClose(stream);
 297
 298     if (stream->client) {
 299         CFStreamClientContext *cbContext;
 300         cbContext = &(stream->client->cbContext);
 301         if (cbContext->info && cbContext->release) {
 302             cbContext->release(cbContext->info);
 303         }
 304         _CFStreamDetachSource(stream);
 305         if (stream->client->runLoopsAndModes) {
 306             CFRelease(stream->client->runLoopsAndModes);
 307         }
 308
 309         CFAllocatorDeallocate(alloc, stream->client);
 310         stream->client = NULL; // Just in case finalize, below, calls back in to us
 311     }
 312     if (cb->finalize) {
 313         if (cb->version == 0) {
 314             ((void(*)(void *))cb->finalize)(_CFStreamGetInfoPointer(stream));
 315         } else {
 316             cb->finalize(stream, _CFStreamGetInfoPointer(stream));
 317         }
 318     }
 319     if (stream->error) {
 320         if (cb->version < 2) {
 321             CFAllocatorDeallocate(alloc, stream->error);
 322         } else {
 323             CFRelease(stream->error);
 324         }
 325     }
 326     if (!__CFBitIsSet(stream->flags, CONSTANT_CALLBACKS)) {
 327         CFAllocatorDeallocate(alloc, (void *)stream->callBacks);
 328     }
 329         if (stream->_reserved1)
 330                 _CFStreamDestroyReserved(alloc, stream->_reserved1);
 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 __private_extern__ 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->_reserved1 = _CFStreamCreateReserved(allocator);
 389     }
 390     return newStream;
 391 }
 392
 393 __private_extern__ struct _CFStream *_CFStreamCreateWithConstantCallbacks(CFAllocatorRef alloc, void *info,  const struct _CFStreamCallBacks *cb, Boolean isReading) {
 394     struct _CFStream *newStream;
 395     if (cb->version != 1) return NULL;
 396     newStream = _CFStreamCreate(alloc, isReading);
 397     if (newStream) {
 398         __CFBitSet(newStream->flags, CONSTANT_CALLBACKS);
 399         newStream->callBacks = cb;
 400         if (cb->create) {
 401             newStream->info = cb->create(newStream, info);
 402         } else {
 403             newStream->info = info;
 404         }
 405     }
 406     return newStream;
 407 }
 408
 409 CF_EXPORT void _CFStreamSetInfoPointer(struct _CFStream *stream, void *info, const struct _CFStreamCallBacks *cb) {
 410     if (info != stream->info) {
 411         if (stream->callBacks->finalize) {
 412             stream->callBacks->finalize(stream, stream->info);
 413         }
 414         if (cb->create) {
 415             stream->info = cb->create(stream, info);
 416         } else {
 417             stream->info = info;
 418         }
 419     }
 420     stream->callBacks = cb;
 421 }
 422
 423
 424 CF_EXPORT CFReadStreamRef CFReadStreamCreate(CFAllocatorRef alloc, const CFReadStreamCallBacks *callbacks, void *info) {
 425     struct _CFStream *newStream = _CFStreamCreate(alloc, TRUE);
 426     struct _CFStreamCallBacks *cb;
 427     if (!newStream) return NULL;
 428     cb = (struct _CFStreamCallBacks *)CFAllocatorAllocate(alloc, sizeof(struct _CFStreamCallBacks), 0);
 429     if (!cb) {
 430         CFRelease(newStream);
 431         return NULL;
 432     }
 433     if (callbacks->version == 0) {
 434         CFReadStreamCallBacksV0 *cbV0 = (CFReadStreamCallBacksV0 *)callbacks;
 435         CFStreamClientContext *ctxt = (CFStreamClientContext *)info;
 436         newStream->info = ctxt->retain ? (void *)ctxt->retain(ctxt->info) : ctxt->info;
 437         cb->version = 0;
 438         cb->create = (void *(*)(struct _CFStream *, void *))ctxt->retain;
 439         cb->finalize = (void(*)(struct _CFStream *, void *))ctxt->release;
 440         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))ctxt->copyDescription;
 441         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))cbV0->open;
 442         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))cbV0->openCompleted;
 443         cb->read = (CFIndex (*)(CFReadStreamRef, UInt8 *, CFIndex, CFErrorRef *, Boolean *, void *))cbV0->read;
 444         cb->getBuffer = (const UInt8 *(*)(CFReadStreamRef, CFIndex, CFIndex *, CFErrorRef *, Boolean *, void *))cbV0->getBuffer;
 445         cb->canRead = (Boolean (*)(CFReadStreamRef, CFErrorRef*, void*))cbV0->canRead;
 446         cb->write = NULL;
 447         cb->canWrite = NULL;
 448         cb->close = (void (*)(struct _CFStream *, void *))cbV0->close;
 449         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))cbV0->copyProperty;
 450         cb->setProperty = NULL;
 451         cb->requestEvents = NULL;
 452         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV0->schedule;
 453         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV0->unschedule;
 454     } else if (callbacks->version == 1) {
 455         CFReadStreamCallBacksV1 *cbV1 = (CFReadStreamCallBacksV1 *)callbacks;
 456         newStream->info = cbV1->create ? cbV1->create((CFReadStreamRef)newStream, info) : info;
 457         cb->version = 1;
 458         cb->create = (void *(*)(struct _CFStream *, void *))cbV1->create;
 459         cb->finalize = (void(*)(struct _CFStream *, void *))cbV1->finalize;
 460         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))cbV1->copyDescription;
 461         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))cbV1->open;
 462         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))cbV1->openCompleted;
 463         cb->read = (CFIndex (*)(CFReadStreamRef, UInt8 *, CFIndex, CFErrorRef *, Boolean *, void *))cbV1->read;
 464         cb->getBuffer = (const UInt8 *(*)(CFReadStreamRef, CFIndex, CFIndex *, CFErrorRef *, Boolean *, void *))cbV1->getBuffer;
 465         cb->canRead = (Boolean (*)(CFReadStreamRef, CFErrorRef*, void*))cbV1->canRead;
 466         cb->write = NULL;
 467         cb->canWrite = NULL;
 468         cb->close = (void (*)(struct _CFStream *, void *))cbV1->close;
 469         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))cbV1->copyProperty;
 470         cb->setProperty = (Boolean(*)(struct _CFStream *, CFStringRef, CFTypeRef, void *))cbV1->setProperty;
 471         cb->requestEvents = (void(*)(struct _CFStream *, CFOptionFlags, void *))cbV1->requestEvents;
 472         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV1->schedule;
 473         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV1->unschedule;
 474     } else {
 475         newStream->info = callbacks->create ? callbacks->create((CFReadStreamRef)newStream, info) : info;
 476         cb->version = 2;
 477         cb->create = (void *(*)(struct _CFStream *, void *))callbacks->create;
 478         cb->finalize = (void(*)(struct _CFStream *, void *))callbacks->finalize;
 479         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))callbacks->copyDescription;
 480         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))callbacks->open;
 481         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))callbacks->openCompleted;
 482         cb->read = callbacks->read;
 483         cb->getBuffer = callbacks->getBuffer;
 484         cb->canRead = callbacks->canRead;
 485         cb->write = NULL;
 486         cb->canWrite = NULL;
 487         cb->close = (void (*)(struct _CFStream *, void *))callbacks->close;
 488         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))callbacks->copyProperty;
 489         cb->setProperty = (Boolean(*)(struct _CFStream *, CFStringRef, CFTypeRef, void *))callbacks->setProperty;
 490         cb->requestEvents = (void(*)(struct _CFStream *, CFOptionFlags, void *))callbacks->requestEvents;
 491         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))callbacks->schedule;
 492         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))callbacks->unschedule;
 493    }
 494
 495     newStream->callBacks = cb;
 496     return (CFReadStreamRef)newStream;
 497 }
 498
 499 CF_EXPORT CFWriteStreamRef CFWriteStreamCreate(CFAllocatorRef alloc, const CFWriteStreamCallBacks *callbacks, void *info) {
 500     struct _CFStream *newStream = _CFStreamCreate(alloc, FALSE);
 501     struct _CFStreamCallBacks *cb;
 502     if (!newStream) return NULL;
 503     cb = (struct _CFStreamCallBacks *)CFAllocatorAllocate(alloc, sizeof(struct _CFStreamCallBacks), 0);
 504     if (!cb) {
 505         CFRelease(newStream);
 506         return NULL;
 507     }
 508     if (callbacks->version == 0) {
 509         CFWriteStreamCallBacksV0 *cbV0 = (CFWriteStreamCallBacksV0 *)callbacks;
 510         CFStreamClientContext *ctxt = (CFStreamClientContext *)info;
 511         newStream->info = ctxt->retain ? (void *)ctxt->retain(ctxt->info) : ctxt->info;
 512         cb->version = 0;
 513         cb->create = (void *(*)(struct _CFStream *, void *))ctxt->retain;
 514         cb->finalize = (void(*)(struct _CFStream *, void *))ctxt->release;
 515         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))ctxt->copyDescription;
 516         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))cbV0->open;
 517         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))cbV0->openCompleted;
 518         cb->read = NULL;
 519         cb->getBuffer = NULL;
 520         cb->canRead = NULL;
 521         cb->write = (CFIndex(*)(CFWriteStreamRef stream, const UInt8 *buffer, CFIndex bufferLength, CFErrorRef *error, void *info))cbV0->write;
 522         cb->canWrite = (Boolean(*)(CFWriteStreamRef stream, CFErrorRef *error, void *info))cbV0->canWrite;
 523         cb->close = (void (*)(struct _CFStream *, void *))cbV0->close;
 524         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))cbV0->copyProperty;
 525         cb->setProperty = NULL;
 526         cb->requestEvents = NULL;
 527         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV0->schedule;
 528         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV0->unschedule;
 529     } else if (callbacks->version == 1) {
 530         CFWriteStreamCallBacksV1 *cbV1 = (CFWriteStreamCallBacksV1 *)callbacks;
 531         cb->version = 1;
 532         newStream->info = cbV1->create ? cbV1->create((CFWriteStreamRef)newStream, info) : info;
 533         cb->create = (void *(*)(struct _CFStream *, void *))cbV1->create;
 534         cb->finalize = (void(*)(struct _CFStream *, void *))cbV1->finalize;
 535         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))cbV1->copyDescription;
 536         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))cbV1->open;
 537         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))cbV1->openCompleted;
 538         cb->read = NULL;
 539         cb->getBuffer = NULL;
 540         cb->canRead = NULL;
 541         cb->write = (CFIndex(*)(CFWriteStreamRef stream, const UInt8 *buffer, CFIndex bufferLength, CFErrorRef *error, void *info))cbV1->write;
 542         cb->canWrite = (Boolean(*)(CFWriteStreamRef stream, CFErrorRef *error, void *info))cbV1->canWrite;
 543         cb->close = (void (*)(struct _CFStream *, void *))cbV1->close;
 544         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))cbV1->copyProperty;
 545         cb->setProperty = (Boolean (*)(struct _CFStream *, CFStringRef, CFTypeRef, void *))cbV1->setProperty;
 546         cb->requestEvents = (void(*)(struct _CFStream *, CFOptionFlags, void *))cbV1->requestEvents;
 547         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV1->schedule;
 548         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))cbV1->unschedule;
 549     } else {
 550         cb->version = callbacks->version;
 551         newStream->info = callbacks->create ? callbacks->create((CFWriteStreamRef)newStream, info) : info;
 552         cb->create = (void *(*)(struct _CFStream *, void *))callbacks->create;
 553         cb->finalize = (void(*)(struct _CFStream *, void *))callbacks->finalize;
 554         cb->copyDescription = (CFStringRef(*)(struct _CFStream *, void *))callbacks->copyDescription;
 555         cb->open = (Boolean(*)(struct _CFStream *, CFErrorRef *, Boolean *, void *))callbacks->open;
 556         cb->openCompleted = (Boolean (*)(struct _CFStream *, CFErrorRef *, void *))callbacks->openCompleted;
 557         cb->read = NULL;
 558         cb->getBuffer = NULL;
 559         cb->canRead = NULL;
 560         cb->write = callbacks->write;
 561         cb->canWrite = callbacks->canWrite;
 562         cb->close = (void (*)(struct _CFStream *, void *))callbacks->close;
 563         cb->copyProperty = (CFTypeRef (*)(struct _CFStream *, CFStringRef, void *))callbacks->copyProperty;
 564         cb->setProperty = (Boolean (*)(struct _CFStream *, CFStringRef, CFTypeRef, void *))callbacks->setProperty;
 565         cb->requestEvents = (void(*)(struct _CFStream *, CFOptionFlags, void *))callbacks->requestEvents;
 566         cb->schedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))callbacks->schedule;
 567         cb->unschedule = (void (*)(struct _CFStream *, CFRunLoopRef, CFStringRef, void *))callbacks->unschedule;
 568     }
 569     newStream->callBacks = cb;
 570     return (CFWriteStreamRef)newStream;
 571 }
 572
 573 static void _signalEventSync(struct _CFStream* stream, CFOptionFlags whatToSignal)
 574 {
 575     CFOptionFlags eventMask;
 576
 577     __CFBitSet(stream->flags, CALLING_CLIENT);
 578
 579     void* info = NULL;
 580     void (*release) (void*) = NULL;
 581
 582     if (stream->client->cbContext.retain == NULL)
 583         info = stream->client->cbContext.info;
 584     else {
 585         info = stream->client->cbContext.retain(stream->client->cbContext.info);
 586         release = stream->client->cbContext.release;
 587     }
 588
 589     for (eventMask = 1; eventMask <= whatToSignal; eventMask = eventMask << 1) {
 590         if ((eventMask & whatToSignal) && (stream->client->when & eventMask)) {
 591             stream->client->cb(stream, eventMask, info);
 592
 593             /* What happens if the callback sets the client to NULL?  We're in a loop here... Hmm. */
 594             /* After writing that comment, I see: <rdar://problem/6793636> CFReadStreamSetClient(..., NULL) unsafely releases info pointer immediately */
 595         /* Of note, when the stream callbacks are set to to NULL, we're re-initalized so as not to receive more events, so we
 596          * should break pout of this loop */
 597         }
 598     }
 599
 600     if (release)
 601         (*release) (info);
 602
 603     __CFBitClear(stream->flags, CALLING_CLIENT);
 604 }
 605
 606 static void _cfstream_solo_signalEventSync(void* info)
 607 {
 608     CFTypeID typeID = CFGetTypeID((CFTypeRef) info);
 609
 610     if (typeID != CFReadStreamGetTypeID() && typeID != CFWriteStreamGetTypeID()) {
 611         CFLog(__kCFLogAssertion, CFSTR("Expected an read or write stream for %p"), info);
 612 #if defined(DEBUG)
 613         abort();
 614 #endif
 615     } else {
 616         struct _CFStream* stream = (struct _CFStream*) info;
 617         CFOptionFlags whatToSignal = stream->client->whatToSignal;
 618         stream->client->whatToSignal = 0;
 619
 620         /* Since the array version holds a retain, we do it here as well, as opposed to taking a second retain in the client callback */
 621         CFRetain(stream);
 622         _signalEventSync(stream, whatToSignal);
 623         CFRelease(stream);
 624     }
 625 }
 626
 627 static void _cfstream_shared_signalEventSync(void* info)
 628 {
 629     CFTypeID typeID = CFGetTypeID((CFTypeRef) info);
 630
 631     if (typeID != CFArrayGetTypeID()) {
 632         CFLog(__kCFLogAssertion, CFSTR("Expected an array for %p"), info);
 633 #if defined(DEBUG)
 634         abort();
 635 #endif
 636     } else {
 637         CFMutableArrayRef list = (CFMutableArrayRef) info;
 638         CFIndex c, i;
 639         CFOptionFlags whatToSignal = 0;
 640         struct _CFStream* stream = NULL;
 641
 642         __CFSpinLock(&sSourceLock);
 643
 644         /* Looks like, we grab the first stream that wants an event... */
 645         /* Note that I grab an extra retain when I pull out the stream here... */
 646         c = CFArrayGetCount(list);
 647         for (i = 0; i < c; i++) {
 648             struct _CFStream* s = (struct _CFStream*)CFArrayGetValueAtIndex(list, i);
 649
 650             if (s->client->whatToSignal) {
 651                 stream = s;
 652                 CFRetain(stream);
 653                 whatToSignal = stream->client->whatToSignal;
 654                 s->client->whatToSignal = 0;
 655                 break;
 656             }
 657         }
 658
 659         /* And then we also signal any other streams in this array so that we get them next go? */
 660         for (; i < c;  i++) {
 661             struct _CFStream* s = (struct _CFStream*)CFArrayGetValueAtIndex(list, i);
 662             if (s->client->whatToSignal) {
 663                         CFRunLoopSourceRef source = _CFStreamCopySource(s);
 664                         if (source) {
 665                                 CFRunLoopSourceSignal(source);
 666                                 CFRelease(source);
 667                         }
 668             break;
 669             }
 670         }
 671
 672         __CFSpinUnlock(&sSourceLock);
 673
 674         /* We're sitting here now, possibly with a stream that needs to be processed by the common routine */
 675         if (stream) {
 676             _signalEventSync(stream, whatToSignal);
 677
 678             /* Lose our extra retain */
 679             CFRelease(stream);
 680         }
 681     }
 682 }
 683
 684 /* This routine is to be considered unsafe... */
 685 static CFArrayRef _CFStreamGetRunLoopsAndModes(struct _CFStream *stream)
 686 {
 687     CFArrayRef result = NULL;
 688     if (stream && stream->client) {
 689         _CFStreamLock(stream);
 690         struct CFStreamAux* aux = _CFStreamGetAuxRecord(stream);
 691         if (aux->previousRunloopsAndModes) {
 692             CFRelease(aux->previousRunloopsAndModes);
 693             aux->previousRunloopsAndModes = NULL;
 694         }
 695         if (stream->client->runLoopsAndModes) {
 696             aux->previousRunloopsAndModes = CFArrayCreateCopy(CFGetAllocator(stream), stream->client->runLoopsAndModes);
 697         }
 698         result = aux->previousRunloopsAndModes;
 699         checkRLMArray(result);
 700         _CFStreamUnlock(stream);
 701     }
 702     return result;
 703 }
 704
 705 static CFArrayRef _CFStreamCopyRunLoopsAndModes(struct _CFStream *stream)
 706 {
 707     CFArrayRef result = NULL;
 708     if (stream && stream->client) {
 709         _CFStreamLock(stream);
 710         if (stream->client->runLoopsAndModes) {
 711             result = CFArrayCreateCopy(CFGetAllocator(stream), stream->client->runLoopsAndModes);
 712         }
 713         checkRLMArray(result);
 714         _CFStreamUnlock(stream);
 715     }
 716     return result;
 717 }
 718
 719 static void _wakeUpRunLoop(struct _CFStream *stream) {
 720     CFArrayRef rlArray = _CFStreamCopyRunLoopsAndModes(stream);
 721     if (rlArray) {
 722         CFIndex cnt = CFArrayGetCount(rlArray);
 723         CFRunLoopRef rl;
 724
 725         if (cnt == 2) {
 726             rl = (CFRunLoopRef)CFArrayGetValueAtIndex(rlArray, 0);
 727         } else if (cnt > 2) {
 728             CFIndex idx;
 729             rl = (CFRunLoopRef)CFArrayGetValueAtIndex(rlArray, 0);
 730             for (idx = 2; NULL != rl && idx < cnt; idx+=2) {
 731                 CFRunLoopRef value = (CFRunLoopRef)CFArrayGetValueAtIndex(rlArray, idx);
 732                 if (value != rl) rl = NULL;
 733             }
 734             if (NULL == rl) {   /* more than one different rl, so we must pick one */
 735                 for (idx = 0; idx < cnt; idx+=2) {
 736                     CFRunLoopRef value = (CFRunLoopRef)CFArrayGetValueAtIndex(rlArray, idx);
 737                     CFStringRef currentMode = CFRunLoopCopyCurrentMode(value);
 738                     if (NULL != currentMode && CFEqual(currentMode, CFArrayGetValueAtIndex(rlArray, idx+1)) && CFRunLoopIsWaiting(value)) {
 739                         CFRelease(currentMode);
 740                         rl = value;
 741                         break;
 742                     }
 743                     if (NULL != currentMode) CFRelease(currentMode);
 744                 }
 745                 if (NULL == rl) {       /* didn't choose one above, so choose first */
 746                     rl = (CFRunLoopRef)CFArrayGetValueAtIndex(rlArray, 0);
 747                 }
 748             }
 749         }
 750         if (NULL != rl && CFRunLoopIsWaiting(rl))
 751             CFRunLoopWakeUp(rl);
 752     }
 753     CFRelease(rlArray);
 754 }
 755
 756 __private_extern__ void _CFStreamSignalEvent(struct _CFStream *stream, CFStreamEventType event, CFErrorRef error, Boolean synchronousAllowed) {
 757     // Update our internal status; we must use the primitive __CFStreamGetStatus(), because CFStreamGetStatus() calls us, and we can end up in an infinite loop.
 758     CFStreamStatus status = __CFStreamGetStatus(stream);
 759
 760     // Sanity check the event
 761     if (status == kCFStreamStatusNotOpen) {
 762         // No events allowed; this is almost certainly a bug in the stream's implementation
 763         CFLog(__kCFLogAssertion, CFSTR("Stream %p is sending an event before being opened"), stream);
 764         event = 0;
 765     } else if (status == kCFStreamStatusClosed || status == kCFStreamStatusError) {
 766         // no further events are allowed
 767         event = 0;
 768     } else if (status == kCFStreamStatusAtEnd) {
 769         // Only error events are allowed
 770         event &= kCFStreamEventErrorOccurred;
 771     } else if (status != kCFStreamStatusOpening) {
 772         // cannot send open completed; that happened already
 773         event &= ~kCFStreamEventOpenCompleted;
 774     }
 775
 776     // Change status if appropriate
 777     if (event & kCFStreamEventOpenCompleted && status == kCFStreamStatusOpening) {
 778         _CFStreamSetStatusCode(stream, kCFStreamStatusOpen);
 779     }
 780     if (event & kCFStreamEventEndEncountered && status < kCFStreamStatusAtEnd) {
 781         _CFStreamSetStatusCode(stream, kCFStreamStatusAtEnd);
 782     }
 783     if (event & kCFStreamEventErrorOccurred) {
 784         if (_CFStreamGetCallBackPtr(stream)->version < 2) {
 785             _CFStreamSetStreamError(stream, (CFStreamError *)error);
 786         } else {
 787             CFAssert(error, __kCFLogAssertion, "CFStream: kCFStreamEventErrorOccurred signalled, but error is NULL!");
 788             CFRetain(error);
 789             if (stream->error) CFRelease(stream->error);
 790             stream->error = error;
 791         }
 792         _CFStreamSetStatusCode(stream, kCFStreamStatusError);
 793     }
 794
 795     // Now signal any pertinent event
 796     if (stream->client && (stream->client->when & event) != 0) {
 797         CFRunLoopSourceRef source = _CFStreamCopySource(stream);
 798
 799         if (source) {
 800             Boolean signalNow = FALSE;
 801
 802             stream->client->whatToSignal |= event;
 803
 804             if (synchronousAllowed && !__CFBitIsSet(stream->flags, CALLING_CLIENT)) {
 805
 806                 CFRunLoopRef rl = CFRunLoopGetCurrent();
 807                 CFStringRef mode = CFRunLoopCopyCurrentMode(rl);
 808
 809                 if (mode) {
 810                     if (CFRunLoopContainsSource(rl, source, mode))
 811                         signalNow = TRUE;
 812                 }
 813                 if (mode)
 814                     CFRelease(mode);
 815             }
 816
 817             if (signalNow) {
 818                 // Can call out safely right now
 819                 _cfstream_solo_signalEventSync(stream);
 820             } else {
 821                 // Schedule for later delivery
 822                 if (source) {
 823                     CFRunLoopSourceSignal(source);
 824                 }
 825                 _wakeUpRunLoop(stream);
 826             }
 827
 828                         CFRelease(source);
 829         }
 830     }
 831 }
 832
 833 __private_extern__ CFStreamStatus _CFStreamGetStatus(struct _CFStream *stream) {
 834   CFStreamStatus status = __CFStreamGetStatus(stream);
 835   // 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.
 836   __CFBitSet(stream->flags, CALLING_CLIENT);
 837   if (status == kCFStreamStatusOpening) {
 838     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
 839     if (cb->openCompleted) {
 840       Boolean isComplete;
 841       if (cb->version < 2) {
 842         CFStreamError err = {0, 0};
 843         isComplete = ((_CFStreamCBOpenCompletedV1)(cb->openCompleted))(stream, &err, _CFStreamGetInfoPointer(stream));
 844         if (err.error != 0) _CFStreamSetStreamError(stream, &err);
 845       } else {
 846         isComplete = cb->openCompleted(stream, &(stream->error), _CFStreamGetInfoPointer(stream));
 847       }
 848       if (isComplete) {
 849         if (!stream->error) {
 850           status = kCFStreamStatusOpen;
 851         } else {
 852           status = kCFStreamStatusError;
 853         }
 854         _CFStreamSetStatusCode(stream, status);
 855         if (status == kCFStreamStatusOpen) {
 856           _CFStreamScheduleEvent(stream, kCFStreamEventOpenCompleted);
 857         } else {
 858           _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
 859         }
 860       }
 861     }
 862   }
 863   __CFBitClear(stream->flags, CALLING_CLIENT);
 864   return status;
 865 }
 866
 867 CF_EXPORT CFStreamStatus CFReadStreamGetStatus(CFReadStreamRef stream) {
 868     CF_OBJC_FUNCDISPATCH0(__kCFReadStreamTypeID, CFStreamStatus, stream, "streamStatus");
 869     return _CFStreamGetStatus((struct _CFStream *)stream);
 870 }
 871
 872 CF_EXPORT CFStreamStatus CFWriteStreamGetStatus(CFWriteStreamRef stream) {
 873     CF_OBJC_FUNCDISPATCH0(__kCFWriteStreamTypeID, CFStreamStatus, stream, "streamStatus");
 874     return _CFStreamGetStatus((struct _CFStream *)stream);
 875 }
 876
 877 static CFStreamError _CFStreamGetStreamError(struct _CFStream *stream) {
 878     CFStreamError result;
 879     if (!stream->error) {
 880         result.error = 0;
 881         result.domain = 0;
 882     } else if (_CFStreamGetCallBackPtr(stream)->version < 2) {
 883         CFStreamError *streamError = (CFStreamError *)(stream->error);
 884         result.error = streamError->error;
 885         result.domain = streamError->domain;
 886     } else {
 887         result = _CFStreamErrorFromError(stream->error);
 888     }
 889     return result;
 890 }
 891
 892 CF_EXPORT CFStreamError CFReadStreamGetError(CFReadStreamRef stream) {
 893     CF_OBJC_FUNCDISPATCH0(__kCFReadStreamTypeID, CFStreamError, stream, "_cfStreamError");
 894     return _CFStreamGetStreamError((struct _CFStream *)stream);
 895 }
 896
 897 CF_EXPORT CFStreamError CFWriteStreamGetError(CFWriteStreamRef stream) {
 898     CF_OBJC_FUNCDISPATCH0(__kCFWriteStreamTypeID, CFStreamError, stream, "_cfStreamError");
 899     return _CFStreamGetStreamError((struct _CFStream *)stream);
 900 }
 901
 902 static CFErrorRef _CFStreamCopyError(struct _CFStream *stream) {
 903     if (!stream->error) {
 904         return NULL;
 905     } else if (_CFStreamGetCallBackPtr(stream)->version < 2) {
 906         return _CFErrorFromStreamError(CFGetAllocator(stream), (CFStreamError *)(stream->error));
 907     } else {
 908         CFRetain(stream->error);
 909         return stream->error;
 910     }
 911 }
 912
 913 CF_EXPORT CFErrorRef CFReadStreamCopyError(CFReadStreamRef stream) {
 914     CF_OBJC_FUNCDISPATCH0(__kCFReadStreamTypeID, CFErrorRef, stream, "streamError");
 915     return _CFStreamCopyError((struct _CFStream *)stream);
 916 }
 917
 918 CF_EXPORT CFErrorRef CFWriteStreamCopyError(CFWriteStreamRef stream) {
 919     return _CFStreamCopyError((struct _CFStream *)stream);
 920     CF_OBJC_FUNCDISPATCH0(__kCFWriteStreamTypeID, CFErrorRef, stream, "streamError");
 921 }
 922
 923 __private_extern__ Boolean _CFStreamOpen(struct _CFStream *stream) {
 924     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
 925     Boolean success, openComplete;
 926     if (_CFStreamGetStatus(stream) != kCFStreamStatusNotOpen) {
 927         return FALSE;
 928     }
 929     __CFBitSet(stream->flags, CALLING_CLIENT);
 930     _CFStreamSetStatusCode(stream, kCFStreamStatusOpening);
 931     if (cb->open) {
 932         if (cb->version < 2) {
 933             CFStreamError err = {0, 0};
 934             success = ((_CFStreamCBOpenV1)(cb->open))(stream, &err, &openComplete, _CFStreamGetInfoPointer(stream));
 935             if (err.error != 0) _CFStreamSetStreamError(stream, &err);
 936         } else {
 937             success = cb->open(stream, &(stream->error), &openComplete, _CFStreamGetInfoPointer(stream));
 938         }
 939     } else {
 940         success = TRUE;
 941         openComplete = TRUE;
 942     }
 943     if (openComplete) {
 944         if (success) {
 945             // 2957690 - Guard against the possibility that the stream has already signalled itself in to a later state (like AtEnd)
 946             if (__CFStreamGetStatus(stream) == kCFStreamStatusOpening) {
 947                 _CFStreamSetStatusCode(stream, kCFStreamStatusOpen);
 948             }
 949             _CFStreamScheduleEvent(stream, kCFStreamEventOpenCompleted);
 950         } else {
 951 #if DEPLOYMENT_TARGET_WINDOWS
 952             _CFStreamClose(stream);
 953 #endif
 954             _CFStreamSetStatusCode(stream, kCFStreamStatusError);
 955             _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
 956         }
 957     }
 958     __CFBitClear(stream->flags, CALLING_CLIENT);
 959     return success;
 960 }
 961
 962 CF_EXPORT Boolean CFReadStreamOpen(CFReadStreamRef stream) {
 963     if(CF_IS_OBJC(__kCFReadStreamTypeID, stream)) {
 964         CF_OBJC_VOIDCALL0(stream, "open");
 965         return TRUE;
 966     }
 967     return _CFStreamOpen((struct _CFStream *)stream);
 968 }
 969
 970 CF_EXPORT Boolean CFWriteStreamOpen(CFWriteStreamRef stream) {
 971     if(CF_IS_OBJC(__kCFWriteStreamTypeID, stream)) {
 972         CF_OBJC_VOIDCALL0(stream, "open");
 973         return TRUE;
 974     }
 975     return _CFStreamOpen((struct _CFStream *)stream);
 976 }
 977
 978 CF_EXPORT void CFReadStreamClose(CFReadStreamRef stream) {
 979     CF_OBJC_FUNCDISPATCH0(__kCFReadStreamTypeID, void, stream, "close");
 980     _CFStreamClose((struct _CFStream *)stream);
 981 }
 982
 983 CF_EXPORT void CFWriteStreamClose(CFWriteStreamRef stream) {
 984     CF_OBJC_FUNCDISPATCH0(__kCFWriteStreamTypeID, void, stream, "close");
 985     _CFStreamClose((struct _CFStream *)stream);
 986 }
 987
 988 CF_EXPORT Boolean CFReadStreamHasBytesAvailable(CFReadStreamRef readStream) {
 989     CF_OBJC_FUNCDISPATCH0(__kCFReadStreamTypeID, Boolean, readStream, "hasBytesAvailable");
 990     struct _CFStream *stream = (struct _CFStream *)readStream;
 991     CFStreamStatus status = _CFStreamGetStatus(stream);
 992     const struct _CFStreamCallBacks *cb;
 993     if (status != kCFStreamStatusOpen && status != kCFStreamStatusReading) {
 994         return FALSE;
 995     }
 996     cb  = _CFStreamGetCallBackPtr(stream);
 997     if (cb->canRead == NULL) {
 998         return TRUE;  // No way to know without trying....
 999     } else {
1000         Boolean result;
1001         __CFBitSet(stream->flags, CALLING_CLIENT);
1002         if (cb->version < 2) {
1003             result = ((_CFStreamCBCanReadV1)(cb->canRead))((CFReadStreamRef)stream, _CFStreamGetInfoPointer(stream));
1004         } else {
1005             result = cb->canRead((CFReadStreamRef)stream, &(stream->error), _CFStreamGetInfoPointer(stream));
1006             if (stream->error) {
1007                 _CFStreamSetStatusCode(stream, kCFStreamStatusError);
1008                 _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
1009             }
1010         }
1011         __CFBitClear(stream->flags, CALLING_CLIENT);
1012         return result;
1013     }
1014 }
1015
1016 static void waitForOpen(struct _CFStream *stream);
1017 CFIndex CFReadStreamRead(CFReadStreamRef readStream, UInt8 *buffer, CFIndex bufferLength) {
1018     CF_OBJC_FUNCDISPATCH2(__kCFReadStreamTypeID, CFIndex, readStream, "read:maxLength:", buffer, bufferLength);
1019     struct _CFStream *stream = (struct _CFStream *)readStream;
1020     CFStreamStatus status = _CFStreamGetStatus(stream);
1021     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1022     if (status == kCFStreamStatusOpening) {
1023         __CFBitSet(stream->flags, CALLING_CLIENT);
1024         waitForOpen(stream);
1025         __CFBitClear(stream->flags, CALLING_CLIENT);
1026         status = _CFStreamGetStatus(stream);
1027     }
1028
1029     if (status != kCFStreamStatusOpen && status != kCFStreamStatusReading && status != kCFStreamStatusAtEnd) {
1030         return -1;
1031     } else  if (status == kCFStreamStatusAtEnd) {
1032         return 0;
1033     } else {
1034         Boolean atEOF;
1035         CFIndex bytesRead;
1036         __CFBitSet(stream->flags, CALLING_CLIENT);
1037         if (stream->client) {
1038             stream->client->whatToSignal &= ~kCFStreamEventHasBytesAvailable;
1039         }
1040         _CFStreamSetStatusCode(stream, kCFStreamStatusReading);
1041         if (cb->version < 2) {
1042             CFStreamError err = {0, 0};
1043             bytesRead = ((_CFStreamCBReadV1)(cb->read))((CFReadStreamRef)stream, buffer, bufferLength, &err, &atEOF, _CFStreamGetInfoPointer(stream));
1044             if (err.error != 0) _CFStreamSetStreamError(stream, &err);
1045         } else {
1046             bytesRead = cb->read((CFReadStreamRef)stream, buffer, bufferLength, &(stream->error), &atEOF, _CFStreamGetInfoPointer(stream));
1047         }
1048         if (stream->error) {
1049             bytesRead = -1;
1050             _CFStreamSetStatusCode(stream, kCFStreamStatusError);
1051             _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
1052         } else if (atEOF) {
1053             _CFStreamSetStatusCode(stream, kCFStreamStatusAtEnd);
1054             _CFStreamScheduleEvent(stream, kCFStreamEventEndEncountered);
1055         } else {
1056             _CFStreamSetStatusCode(stream, kCFStreamStatusOpen);
1057         }
1058         __CFBitClear(stream->flags, CALLING_CLIENT);
1059         return bytesRead;
1060     }
1061 }
1062
1063 CF_EXPORT const UInt8 *CFReadStreamGetBuffer(CFReadStreamRef readStream, CFIndex maxBytesToRead, CFIndex *numBytesRead) {
1064     if (CF_IS_OBJC(__kCFReadStreamTypeID, readStream)) {
1065         uint8_t *bufPtr = NULL;
1066         Boolean gotBytes;
1067         CF_OBJC_CALL2(Boolean, gotBytes, readStream, "getBuffer:length:", &bufPtr, numBytesRead);
1068         if(gotBytes) {
1069             return (const UInt8 *)bufPtr;
1070         } else {
1071             return NULL;
1072         }
1073     }
1074     struct _CFStream *stream = (struct _CFStream *)readStream;
1075     CFStreamStatus status = _CFStreamGetStatus(stream);
1076     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1077     const UInt8 *buffer;
1078     if (status == kCFStreamStatusOpening) {
1079         __CFBitSet(stream->flags, CALLING_CLIENT);
1080         waitForOpen(stream);
1081         __CFBitClear(stream->flags, CALLING_CLIENT);
1082         status = _CFStreamGetStatus(stream);
1083     }
1084     if (status != kCFStreamStatusOpen && status != kCFStreamStatusReading && status != kCFStreamStatusAtEnd) {
1085         *numBytesRead = -1;
1086         buffer = NULL;
1087     } else  if (status == kCFStreamStatusAtEnd || cb->getBuffer == NULL) {
1088         *numBytesRead = 0;
1089         buffer = NULL;
1090     } else {
1091         Boolean atEOF;
1092         Boolean hadBytes = stream->client && (stream->client->whatToSignal & kCFStreamEventHasBytesAvailable);
1093         __CFBitSet(stream->flags, CALLING_CLIENT);
1094         if (hadBytes) {
1095             stream->client->whatToSignal &= ~kCFStreamEventHasBytesAvailable;
1096         }
1097         _CFStreamSetStatusCode(stream, kCFStreamStatusReading);
1098         if (cb->version < 2) {
1099             CFStreamError err = {0, 0};
1100             buffer = ((_CFStreamCBGetBufferV1)(cb->getBuffer))((CFReadStreamRef)stream, maxBytesToRead, numBytesRead, &err, &atEOF, _CFStreamGetInfoPointer(stream));
1101             if (err.error != 0) _CFStreamSetStreamError(stream, &err);
1102         } else {
1103             buffer = cb->getBuffer((CFReadStreamRef)stream, maxBytesToRead, numBytesRead, &(stream->error), &atEOF, _CFStreamGetInfoPointer(stream));
1104         }
1105         if (stream->error) {
1106             *numBytesRead = -1;
1107             _CFStreamSetStatusCode(stream, kCFStreamStatusError);
1108             buffer = NULL;
1109             _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
1110         } else if (atEOF) {
1111             _CFStreamSetStatusCode(stream, kCFStreamStatusAtEnd);
1112             _CFStreamScheduleEvent(stream, kCFStreamEventEndEncountered);
1113         } else {
1114             if (!buffer && hadBytes) {
1115                 stream->client->whatToSignal |= kCFStreamEventHasBytesAvailable;
1116             }
1117             _CFStreamSetStatusCode(stream, kCFStreamStatusOpen);
1118         }
1119         __CFBitClear(stream->flags, CALLING_CLIENT);
1120     }
1121     return buffer;
1122 }
1123
1124 CF_EXPORT Boolean CFWriteStreamCanAcceptBytes(CFWriteStreamRef writeStream) {
1125     CF_OBJC_FUNCDISPATCH0(__kCFWriteStreamTypeID, Boolean, writeStream, "hasSpaceAvailable");
1126     struct _CFStream *stream = (struct _CFStream *)writeStream;
1127     CFStreamStatus status = _CFStreamGetStatus(stream);
1128     const struct _CFStreamCallBacks *cb;
1129     if (status != kCFStreamStatusOpen && status != kCFStreamStatusWriting) {
1130         return FALSE;
1131     }
1132     cb  = _CFStreamGetCallBackPtr(stream);
1133     if (cb->canWrite == NULL) {
1134         return TRUE;  // No way to know without trying....
1135     } else {
1136         Boolean result;
1137         __CFBitSet(stream->flags, CALLING_CLIENT);
1138         if (cb->version < 2) {
1139             result = ((_CFStreamCBCanWriteV1)(cb->canWrite))((CFWriteStreamRef)stream, _CFStreamGetInfoPointer(stream));
1140         } else {
1141             result = cb->canWrite((CFWriteStreamRef)stream, &(stream->error), _CFStreamGetInfoPointer(stream));
1142             if (stream->error) {
1143                 _CFStreamSetStatusCode(stream, kCFStreamStatusError);
1144                 _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
1145             }
1146         }
1147         __CFBitClear(stream->flags, CALLING_CLIENT);
1148         return result;
1149     }
1150 }
1151
1152 CF_EXPORT CFIndex CFWriteStreamWrite(CFWriteStreamRef writeStream, const UInt8 *buffer, CFIndex bufferLength) {
1153     CF_OBJC_FUNCDISPATCH2(__kCFWriteStreamTypeID, CFIndex, writeStream, "write:maxLength:", buffer, bufferLength);
1154     struct _CFStream *stream = (struct _CFStream *)writeStream;
1155     CFStreamStatus status = _CFStreamGetStatus(stream);
1156     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1157     if (status == kCFStreamStatusOpening) {
1158         __CFBitSet(stream->flags, CALLING_CLIENT);
1159         waitForOpen(stream);
1160         __CFBitClear(stream->flags, CALLING_CLIENT);
1161         status = _CFStreamGetStatus(stream);
1162     }
1163     if (status != kCFStreamStatusOpen && status != kCFStreamStatusWriting) {
1164         return -1;
1165     } else {
1166         CFIndex result;
1167         __CFBitSet(stream->flags, CALLING_CLIENT);
1168         _CFStreamSetStatusCode(stream, kCFStreamStatusWriting);
1169         if (stream->client) {
1170             stream->client->whatToSignal &= ~kCFStreamEventCanAcceptBytes;
1171         }
1172         if (cb->version < 2) {
1173             CFStreamError err = {0, 0};
1174             result = ((_CFStreamCBWriteV1)(cb->write))((CFWriteStreamRef)stream, buffer, bufferLength, &err, _CFStreamGetInfoPointer(stream));
1175             if (err.error) _CFStreamSetStreamError(stream, &err);
1176         } else {
1177             result = cb->write((CFWriteStreamRef)stream, buffer, bufferLength, &(stream->error), _CFStreamGetInfoPointer(stream));
1178         }
1179         if (stream->error) {
1180             _CFStreamSetStatusCode(stream, kCFStreamStatusError);
1181             _CFStreamScheduleEvent(stream, kCFStreamEventErrorOccurred);
1182         } else if (result == 0) {
1183             _CFStreamSetStatusCode(stream, kCFStreamStatusAtEnd);
1184             _CFStreamScheduleEvent(stream, kCFStreamEventEndEncountered);
1185         } else {
1186             _CFStreamSetStatusCode(stream, kCFStreamStatusOpen);
1187         }
1188         __CFBitClear(stream->flags, CALLING_CLIENT);
1189         return result;
1190     }
1191 }
1192
1193 __private_extern__ CFTypeRef _CFStreamCopyProperty(struct _CFStream *stream, CFStringRef propertyName) {
1194     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1195     if (cb->copyProperty == NULL) {
1196         return NULL;
1197     } else {
1198         CFTypeRef result;
1199         __CFBitSet(stream->flags, CALLING_CLIENT);
1200         result = cb->copyProperty(stream, propertyName, _CFStreamGetInfoPointer(stream));
1201         __CFBitClear(stream->flags, CALLING_CLIENT);
1202         return result;
1203     }
1204 }
1205
1206 CF_EXPORT CFTypeRef CFReadStreamCopyProperty(CFReadStreamRef stream, CFStringRef propertyName) {
1207     CF_OBJC_FUNCDISPATCH1(__kCFReadStreamTypeID, CFTypeRef, stream, "propertyForKey:", propertyName);
1208     return _CFStreamCopyProperty((struct _CFStream *)stream, propertyName);
1209 }
1210
1211 CF_EXPORT CFTypeRef CFWriteStreamCopyProperty(CFWriteStreamRef stream, CFStringRef propertyName) {
1212     CF_OBJC_FUNCDISPATCH1(__kCFWriteStreamTypeID, CFTypeRef, stream, "propertyForKey:", propertyName);
1213     return _CFStreamCopyProperty((struct _CFStream *)stream, propertyName);
1214 }
1215
1216 __private_extern__ Boolean _CFStreamSetProperty(struct _CFStream *stream, CFStringRef prop, CFTypeRef val) {
1217     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1218     if (cb->setProperty == NULL) {
1219         return FALSE;
1220     } else {
1221         Boolean result;
1222         __CFBitSet(stream->flags, CALLING_CLIENT);
1223         result = cb->setProperty(stream, prop, val, _CFStreamGetInfoPointer(stream));
1224         __CFBitClear(stream->flags, CALLING_CLIENT);
1225         return result;
1226     }
1227 }
1228
1229 CF_EXPORT
1230 Boolean CFReadStreamSetProperty(CFReadStreamRef stream, CFStringRef propertyName, CFTypeRef propertyValue) {
1231     CF_OBJC_FUNCDISPATCH2(__kCFReadStreamTypeID, Boolean, stream, "setProperty:forKey:", propertyValue, propertyName);
1232     return _CFStreamSetProperty((struct _CFStream *)stream, propertyName, propertyValue);
1233 }
1234
1235 CF_EXPORT
1236 Boolean CFWriteStreamSetProperty(CFWriteStreamRef stream, CFStringRef propertyName, CFTypeRef propertyValue) {
1237     CF_OBJC_FUNCDISPATCH2(__kCFWriteStreamTypeID, Boolean, stream, "setProperty:forKey:", propertyValue, propertyName);
1238     return _CFStreamSetProperty((struct _CFStream *)stream, propertyName, propertyValue);
1239 }
1240
1241 static void _initializeClient(struct _CFStream *stream) {
1242     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1243     if (!cb->schedule) return; // Do we wish to allow this?
1244     stream->client = (struct _CFStreamClient *)CFAllocatorAllocate(CFGetAllocator(stream), sizeof(struct _CFStreamClient), 0);
1245     memset(stream->client, 0, sizeof(struct _CFStreamClient));
1246 }
1247
1248 /* If we add a setClient callback to the concrete stream callbacks, we must set/clear CALLING_CLIENT around it */
1249 __private_extern__ Boolean _CFStreamSetClient(struct _CFStream *stream, CFOptionFlags streamEvents, void (*clientCB)(struct _CFStream *, CFStreamEventType, void *), CFStreamClientContext *clientCallBackContext) {
1250
1251     Boolean removingClient = (streamEvents == kCFStreamEventNone || clientCB == NULL || clientCallBackContext == NULL);
1252
1253     if (removingClient) {
1254         clientCB = NULL;
1255         streamEvents = kCFStreamEventNone;
1256         clientCallBackContext = NULL;
1257     }
1258     if (!stream->client) {
1259         if (removingClient) {
1260             // We have no client now, and we've been asked to add none???
1261             return TRUE;
1262         }
1263         _initializeClient(stream);
1264         if (!stream->client) {
1265             // Asynch not supported
1266             return FALSE;
1267         }
1268     }
1269     if (stream->client->cb && stream->client->cbContext.release) {
1270         stream->client->cbContext.release(stream->client->cbContext.info);
1271     }
1272     stream->client->cb = clientCB;
1273     if (clientCallBackContext) {
1274         stream->client->cbContext.version = clientCallBackContext->version;
1275         stream->client->cbContext.retain = clientCallBackContext->retain;
1276         stream->client->cbContext.release = clientCallBackContext->release;
1277         stream->client->cbContext.copyDescription = clientCallBackContext->copyDescription;
1278         stream->client->cbContext.info = (clientCallBackContext->retain && clientCallBackContext->info) ? clientCallBackContext->retain(clientCallBackContext->info) : clientCallBackContext->info;
1279     } else {
1280         stream->client->cbContext.retain = NULL;
1281         stream->client->cbContext.release = NULL;
1282         stream->client->cbContext.copyDescription = NULL;
1283         stream->client->cbContext.info = NULL;
1284     }
1285     if (stream->client->when != streamEvents) {
1286         const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1287         stream->client->when = streamEvents;
1288         if (cb->requestEvents) {
1289             cb->requestEvents(stream, streamEvents, _CFStreamGetInfoPointer(stream));
1290         }
1291     }
1292     return TRUE;
1293 }
1294
1295 CF_EXPORT Boolean CFReadStreamSetClient(CFReadStreamRef readStream, CFOptionFlags streamEvents, CFReadStreamClientCallBack clientCB, CFStreamClientContext *clientContext) {
1296     CF_OBJC_FUNCDISPATCH3(__kCFReadStreamTypeID, Boolean, readStream, "_setCFClientFlags:callback:context:", streamEvents, clientCB, clientContext);
1297     streamEvents &= ~kCFStreamEventCanAcceptBytes;
1298     return _CFStreamSetClient((struct _CFStream *)readStream, streamEvents, (void (*)(struct _CFStream *, CFStreamEventType, void *))clientCB, clientContext);
1299 }
1300
1301 CF_EXPORT Boolean CFWriteStreamSetClient(CFWriteStreamRef writeStream, CFOptionFlags streamEvents, CFWriteStreamClientCallBack clientCB, CFStreamClientContext *clientContext) {
1302     CF_OBJC_FUNCDISPATCH3(__kCFWriteStreamTypeID, Boolean, writeStream, "_setCFClientFlags:callback:context:", streamEvents, clientCB, clientContext);
1303     streamEvents &= ~kCFStreamEventHasBytesAvailable;
1304     return _CFStreamSetClient((struct _CFStream *)writeStream, streamEvents, (void (*)(struct _CFStream *, CFStreamEventType, void *))clientCB, clientContext);
1305 }
1306
1307 CF_INLINE void *_CFStreamGetClient(struct _CFStream *stream) {
1308     if (stream->client) return stream->client->cbContext.info;
1309     else return NULL;
1310 }
1311
1312 CF_EXPORT void *_CFReadStreamGetClient(CFReadStreamRef readStream) {
1313     return _CFStreamGetClient((struct _CFStream *)readStream);
1314 }
1315
1316 CF_EXPORT void *_CFWriteStreamGetClient(CFWriteStreamRef writeStream) {
1317     return _CFStreamGetClient((struct _CFStream *)writeStream);
1318 }
1319
1320
1321 __private_extern__ void _CFStreamScheduleWithRunLoop(struct _CFStream *stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1322     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1323
1324     if (! stream->client) {
1325         _initializeClient(stream);
1326         if (!stream->client) return; // we don't support asynch.
1327     }
1328
1329     if (! stream->client->rlSource) {
1330         /* No source, so we join the shared source group */
1331         CFTypeRef a[] = { runLoop, runLoopMode };
1332
1333         CFArrayRef runLoopAndSourceKey = CFArrayCreate(kCFAllocatorSystemDefault, a, sizeof(a) / sizeof(a[0]), &kCFTypeArrayCallBacks);
1334
1335         __CFSpinLock(&sSourceLock);
1336
1337         if (!sSharedSources)
1338             sSharedSources = CFDictionaryCreateMutable(kCFAllocatorSystemDefault, 0, &kCFTypeDictionaryKeyCallBacks, &kCFTypeDictionaryValueCallBacks);
1339
1340         CFMutableArrayRef listOfStreamsSharingASource = (CFMutableArrayRef)CFDictionaryGetValue(sSharedSources, runLoopAndSourceKey);
1341         if (listOfStreamsSharingASource) {
1342                         struct _CFStream* aStream = (struct _CFStream*) CFArrayGetValueAtIndex(listOfStreamsSharingASource, 0);
1343                         CFRunLoopSourceRef source = _CFStreamCopySource(aStream);
1344                         if (source) {
1345                                 _CFStreamSetSource(stream, source, FALSE);
1346                                 CFRelease(source);
1347                         }
1348             CFRetain(listOfStreamsSharingASource);
1349         }
1350         else {
1351             CFRunLoopSourceContext ctxt = {
1352                 0,
1353                 NULL,
1354                 CFRetain,
1355                 CFRelease,
1356                 (CFStringRef(*)(const void *))CFCopyDescription,
1357                 NULL,
1358                 NULL,
1359                 NULL,
1360                 NULL,
1361                 (void(*)(void *))_cfstream_shared_signalEventSync
1362             };
1363
1364             listOfStreamsSharingASource = CFArrayCreateMutable(kCFAllocatorSystemDefault, 0, &kCFTypeArrayCallBacks);
1365             CFDictionaryAddValue(sSharedSources, runLoopAndSourceKey, listOfStreamsSharingASource);
1366
1367             ctxt.info = listOfStreamsSharingASource;
1368
1369                         CFRunLoopSourceRef source = CFRunLoopSourceCreate(kCFAllocatorSystemDefault, 0, &ctxt);
1370                         _CFStreamSetSource(stream, source, FALSE);
1371             CFRunLoopAddSource(runLoop, source, runLoopMode);
1372                         CFRelease(source);
1373         }
1374
1375         CFArrayAppendValue(listOfStreamsSharingASource, stream);
1376         CFDictionaryAddValue(sSharedSources, stream, runLoopAndSourceKey);
1377
1378         CFRelease(runLoopAndSourceKey);
1379         CFRelease(listOfStreamsSharingASource);
1380
1381         __CFBitSet(stream->flags, SHARED_SOURCE);
1382
1383         __CFSpinUnlock(&sSourceLock);
1384     }
1385     else if (__CFBitIsSet(stream->flags, SHARED_SOURCE)) {
1386         /* We were sharing, but now we'll get our own source */
1387
1388         CFArrayRef runLoopAndSourceKey;
1389         CFMutableArrayRef listOfStreamsSharingASource;
1390         CFIndex count, i;
1391
1392         CFAllocatorRef alloc = CFGetAllocator(stream);
1393         CFRunLoopSourceContext ctxt = {
1394             0,
1395             (void *)stream,
1396             NULL,                                                                                                               // Do not use CFRetain/CFRelease callbacks here; that will cause a retain loop
1397             NULL,                                                                                                               // Do not use CFRetain/CFRelease callbacks here; that will cause a retain loop
1398             (CFStringRef(*)(const void *))CFCopyDescription,
1399             NULL,
1400             NULL,
1401             NULL,
1402             NULL,
1403             (void(*)(void *))_cfstream_solo_signalEventSync
1404         };
1405
1406         __CFSpinLock(&sSourceLock);
1407
1408         runLoopAndSourceKey = (CFArrayRef)CFRetain((CFTypeRef)CFDictionaryGetValue(sSharedSources, stream));
1409         listOfStreamsSharingASource = (CFMutableArrayRef)CFDictionaryGetValue(sSharedSources, runLoopAndSourceKey);
1410
1411         count = CFArrayGetCount(listOfStreamsSharingASource);
1412         i = CFArrayGetFirstIndexOfValue(listOfStreamsSharingASource, CFRangeMake(0, count), stream);
1413         if (i != kCFNotFound) {
1414             CFArrayRemoveValueAtIndex(listOfStreamsSharingASource, i);
1415             count--;
1416         }
1417
1418         if (count == 0) {
1419                         CFRunLoopSourceRef source = _CFStreamCopySource(stream);
1420                         if (source) {
1421                                 CFRunLoopRemoveSource((CFRunLoopRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 0), source, (CFStringRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 1));
1422                                 CFRelease(source);
1423                         }
1424             CFDictionaryRemoveValue(sSharedSources, runLoopAndSourceKey);
1425         }
1426
1427         CFDictionaryRemoveValue(sSharedSources, stream);
1428
1429                 _CFStreamSetSource(stream, NULL, count == 0);
1430
1431         __CFBitClear(stream->flags, SHARED_SOURCE);
1432
1433         __CFSpinUnlock(&sSourceLock);
1434
1435                 CFRunLoopSourceRef source = CFRunLoopSourceCreate(alloc, 0, &ctxt);
1436                 _CFStreamSetSource(stream, source, FALSE);
1437         CFRunLoopAddSource((CFRunLoopRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 0), source, (CFStringRef)CFArrayGetValueAtIndex(runLoopAndSourceKey, 1));
1438         CFRelease(runLoopAndSourceKey);
1439
1440         CFRunLoopAddSource(runLoop, source, runLoopMode);
1441
1442                 CFRelease(source);
1443     } else {
1444         /* We're not sharing, so just add the source to the rl & mode */
1445                 CFRunLoopSourceRef source = _CFStreamCopySource(stream);
1446                 if (source) {
1447                         CFRunLoopAddSource(runLoop, source, runLoopMode);
1448                         CFRelease(source);
1449                 }
1450     }
1451
1452     _CFStreamLock(stream);
1453     if (!stream->client->runLoopsAndModes) {
1454         stream->client->runLoopsAndModes = CFArrayCreateMutable(CFGetAllocator(stream), 0, &kCFTypeArrayCallBacks);
1455     }
1456     CFArrayAppendValue(stream->client->runLoopsAndModes, runLoop);
1457     CFArrayAppendValue(stream->client->runLoopsAndModes, runLoopMode);
1458     checkRLMArray(stream->client->runLoopsAndModes);
1459     _CFStreamUnlock(stream);
1460
1461     if (cb->schedule) {
1462         __CFBitSet(stream->flags, CALLING_CLIENT);
1463         cb->schedule(stream, runLoop, runLoopMode, _CFStreamGetInfoPointer(stream));
1464         __CFBitClear(stream->flags, CALLING_CLIENT);
1465     }
1466
1467     /*
1468      * If we've got events pending, we need to wake up and signal
1469      */
1470     if (stream->client && stream->client->whatToSignal != 0) {
1471                 CFRunLoopSourceRef source = _CFStreamCopySource(stream);
1472                 if (source) {
1473                         CFRunLoopSourceSignal(source);
1474                         CFRelease(source);
1475             _wakeUpRunLoop(stream);
1476         }
1477     }
1478 }
1479
1480 CF_EXPORT void CFReadStreamScheduleWithRunLoop(CFReadStreamRef stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1481     CF_OBJC_FUNCDISPATCH2(__kCFReadStreamTypeID, void, stream, "_scheduleInCFRunLoop:forMode:", runLoop, runLoopMode);
1482     _CFStreamScheduleWithRunLoop((struct _CFStream *)stream, runLoop, runLoopMode);
1483 }
1484
1485 CF_EXPORT void CFWriteStreamScheduleWithRunLoop(CFWriteStreamRef stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1486     CF_OBJC_FUNCDISPATCH2(__kCFWriteStreamTypeID, void, stream, "_scheduleInCFRunLoop:forMode:", runLoop, runLoopMode);
1487     _CFStreamScheduleWithRunLoop((struct _CFStream *)stream, runLoop, runLoopMode);
1488 }
1489
1490
1491 __private_extern__ void _CFStreamUnscheduleFromRunLoop(struct _CFStream *stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1492     const struct _CFStreamCallBacks *cb = _CFStreamGetCallBackPtr(stream);
1493     if (!stream->client) return;
1494     if (!stream->client->rlSource) return;
1495
1496     if (!__CFBitIsSet(stream->flags, SHARED_SOURCE)) {
1497                 CFRunLoopSourceRef source = _CFStreamCopySource(stream);
1498                 if (source) {
1499                         CFRunLoopRemoveSource(runLoop, source, runLoopMode);
1500                         CFRelease(source);
1501                 }
1502     } else {
1503         CFArrayRef runLoopAndSourceKey;
1504         CFMutableArrayRef list;
1505         CFIndex count, i;
1506
1507         __CFSpinLock(&sSourceLock);
1508
1509         runLoopAndSourceKey = (CFArrayRef)CFDictionaryGetValue(sSharedSources, stream);
1510         list = (CFMutableArrayRef)CFDictionaryGetValue(sSharedSources, runLoopAndSourceKey);
1511
1512         count = CFArrayGetCount(list);
1513         i = CFArrayGetFirstIndexOfValue(list, CFRangeMake(0, count), stream);
1514         if (i != kCFNotFound) {
1515             CFArrayRemoveValueAtIndex(list, i);
1516             count--;
1517         }
1518
1519         if (count == 0) {
1520                         CFRunLoopSourceRef source = _CFStreamCopySource(stream);
1521                         if (source) {
1522                                 CFRunLoopRemoveSource(runLoop, source, runLoopMode);
1523                                 CFRelease(source);
1524                         }
1525             CFDictionaryRemoveValue(sSharedSources, runLoopAndSourceKey);
1526         }
1527
1528         CFDictionaryRemoveValue(sSharedSources, stream);
1529
1530         _CFStreamSetSource(stream, NULL, count == 0);
1531
1532         __CFBitClear(stream->flags, SHARED_SOURCE);
1533
1534         __CFSpinUnlock(&sSourceLock);
1535     }
1536
1537     _CFStreamLock(stream);
1538     _CFStreamRemoveRunLoopAndModeFromArray(stream->client->runLoopsAndModes, runLoop, runLoopMode);
1539     checkRLMArray(stream->client->runLoopsAndModes);
1540     _CFStreamUnlock(stream);
1541
1542     if (cb->unschedule) {
1543         cb->unschedule(stream, runLoop, runLoopMode, _CFStreamGetInfoPointer(stream));
1544     }
1545 }
1546
1547 CF_EXPORT void CFReadStreamUnscheduleFromRunLoop(CFReadStreamRef stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1548     CF_OBJC_FUNCDISPATCH2(__kCFReadStreamTypeID, void, stream, "_unscheduleFromCFRunLoop:forMode:", runLoop, runLoopMode);
1549     _CFStreamUnscheduleFromRunLoop((struct _CFStream *)stream, runLoop, runLoopMode);
1550 }
1551
1552 void CFWriteStreamUnscheduleFromRunLoop(CFWriteStreamRef stream, CFRunLoopRef runLoop, CFStringRef runLoopMode) {
1553     CF_OBJC_FUNCDISPATCH2(__kCFWriteStreamTypeID, void, stream, "_unscheduleFromCFRunLoop:forMode:", runLoop, runLoopMode);
1554     _CFStreamUnscheduleFromRunLoop((struct _CFStream *)stream, runLoop, runLoopMode);
1555 }
1556
1557 static void waitForOpen(struct _CFStream *stream) {
1558     CFRunLoopRef runLoop = CFRunLoopGetCurrent();
1559     CFStringRef privateMode = CFSTR("_kCFStreamBlockingOpenMode");
1560     _CFStreamScheduleWithRunLoop(stream, runLoop, privateMode);
1561     // 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....
1562     while (__CFStreamGetStatus(stream) == kCFStreamStatusOpening) {
1563         CFRunLoopRunInMode(privateMode, 1e+20, TRUE);
1564     }
1565     _CFStreamUnscheduleFromRunLoop(stream, runLoop, privateMode);
1566 }
1567
1568 CF_EXPORT CFArrayRef _CFReadStreamGetRunLoopsAndModes(CFReadStreamRef readStream) {
1569     return _CFStreamGetRunLoopsAndModes((struct _CFStream *)readStream);
1570 }
1571
1572 CF_EXPORT CFArrayRef _CFWriteStreamGetRunLoopsAndModes(CFWriteStreamRef writeStream) {
1573     return _CFStreamGetRunLoopsAndModes((struct _CFStream *)writeStream);
1574 }
1575
1576 CF_EXPORT CFArrayRef _CFReadStreamCopyRunLoopsAndModes(CFReadStreamRef readStream) {
1577     return _CFStreamCopyRunLoopsAndModes((struct _CFStream *)readStream);
1578 }
1579
1580 CF_EXPORT CFArrayRef _CFWriteStreamCopyRunLoopsAndModes(CFWriteStreamRef writeStream) {
1581     return _CFStreamCopyRunLoopsAndModes((struct _CFStream *)writeStream);
1582 }
1583
1584 CF_EXPORT void CFReadStreamSignalEvent(CFReadStreamRef stream, CFStreamEventType event, const void *error) {
1585     _CFStreamSignalEvent((struct _CFStream *)stream, event, (CFErrorRef)error, TRUE);
1586 }
1587
1588 CF_EXPORT void CFWriteStreamSignalEvent(CFWriteStreamRef stream, CFStreamEventType event, const void *error) {
1589     _CFStreamSignalEvent((struct _CFStream *)stream, event, (CFErrorRef)error, TRUE);
1590 }
1591
1592 CF_EXPORT void _CFReadStreamSignalEventDelayed(CFReadStreamRef stream, CFStreamEventType event, const void *error) {
1593     _CFStreamSignalEvent((struct _CFStream *)stream, event, (CFErrorRef)error, FALSE);
1594 }
1595
1596 CF_EXPORT void _CFReadStreamClearEvent(CFReadStreamRef readStream, CFStreamEventType event) {
1597     struct _CFStream *stream = (struct _CFStream *)readStream;
1598     if (stream->client) {
1599         stream->client->whatToSignal &= ~event;
1600     }
1601 }
1602
1603 CF_EXPORT void _CFWriteStreamSignalEventDelayed(CFWriteStreamRef stream, CFStreamEventType event, const void *error) {
1604     _CFStreamSignalEvent((struct _CFStream *)stream, event, (CFErrorRef)error, FALSE);
1605 }
1606
1607 CF_EXPORT void *CFReadStreamGetInfoPointer(CFReadStreamRef stream) {
1608     return _CFStreamGetInfoPointer((struct _CFStream *)stream);
1609 }
1610
1611 CF_EXPORT void *CFWriteStreamGetInfoPointer(CFWriteStreamRef stream) {
1612     return _CFStreamGetInfoPointer((struct _CFStream *)stream);
1613 }
1614
1615 /* CF_EXPORT */
1616 void _CFStreamSourceScheduleWithRunLoop(CFRunLoopSourceRef source, CFMutableArrayRef runLoopsAndModes, CFRunLoopRef runLoop, CFStringRef runLoopMode)
1617 {
1618     CFIndex count;
1619     CFRange range;
1620
1621     checkRLMArray(runLoopsAndModes);
1622
1623     count = CFArrayGetCount(runLoopsAndModes);
1624     range = CFRangeMake(0, count);
1625
1626     while (range.length) {
1627
1628         CFIndex i = CFArrayGetFirstIndexOfValue(runLoopsAndModes, range, runLoop);
1629
1630         if (i == kCFNotFound)
1631             break;
1632
1633         if (CFEqual(CFArrayGetValueAtIndex(runLoopsAndModes, i + 1), runLoopMode))
1634             return;
1635
1636         range.location = i + 2;
1637         range.length = count - range.location;
1638     }
1639
1640         // Add the new values.
1641     CFArrayAppendValue(runLoopsAndModes, runLoop);
1642     CFArrayAppendValue(runLoopsAndModes, runLoopMode);
1643
1644         // Schedule the source on the new loop and mode.
1645     if (source)
1646         CFRunLoopAddSource(runLoop, source, runLoopMode);
1647 }
1648
1649
1650 /* CF_EXPORT */
1651 void _CFStreamSourceUnscheduleFromRunLoop(CFRunLoopSourceRef source, CFMutableArrayRef runLoopsAndModes, CFRunLoopRef runLoop, CFStringRef runLoopMode)
1652 {
1653     CFIndex count;
1654         CFRange range;
1655
1656     count = CFArrayGetCount(runLoopsAndModes);
1657     range = CFRangeMake(0, count);
1658
1659     checkRLMArray(runLoopsAndModes);
1660
1661         while (range.length) {
1662
1663                 CFIndex i = CFArrayGetFirstIndexOfValue(runLoopsAndModes, range, runLoop);
1664
1665                 // If not found, it's not scheduled on it.
1666                 if (i == kCFNotFound)
1667                         return;
1668
1669                 // Make sure it is scheduled in this mode.
1670                 if (CFEqual(CFArrayGetValueAtIndex(runLoopsAndModes, i + 1), runLoopMode)) {
1671
1672                         // Remove mode and runloop from the list.
1673             CFArrayReplaceValues(runLoopsAndModes, CFRangeMake(i, 2), NULL, 0);
1674
1675             // Remove it from the runloop.
1676             if (source)
1677                 CFRunLoopRemoveSource(runLoop, source, runLoopMode);
1678
1679                         return;
1680                 }
1681
1682         range.location = i + 2;
1683         range.length = count - range.location;
1684         }
1685 }
1686
1687
1688 /* CF_EXPORT */
1689 void _CFStreamSourceScheduleWithAllRunLoops(CFRunLoopSourceRef source, CFArrayRef runLoopsAndModes)
1690 {
1691     CFIndex i, count = CFArrayGetCount(runLoopsAndModes);
1692
1693     if (!source)
1694         return;
1695
1696     checkRLMArray(runLoopsAndModes);
1697
1698     for (i = 0; i < count; i += 2) {
1699
1700         // Make sure it's scheduled on all the right loops and modes.
1701         // Go through the array adding the source to all loops and modes.
1702         CFRunLoopAddSource((CFRunLoopRef)CFArrayGetValueAtIndex(runLoopsAndModes, i),
1703                             source,
1704                             (CFStringRef)CFArrayGetValueAtIndex(runLoopsAndModes, i + 1));
1705         }
1706 }
1707
1708
1709 /* CF_EXPORT */
1710 void _CFStreamSourceUncheduleFromAllRunLoops(CFRunLoopSourceRef source, CFArrayRef runLoopsAndModes)
1711 {
1712     CFIndex i, count = CFArrayGetCount(runLoopsAndModes);
1713
1714     if (!source)
1715         return;
1716
1717     checkRLMArray(runLoopsAndModes);
1718
1719     for (i = 0; i < count; i += 2) {
1720
1721         // Go through the array removing the source from all loops and modes.
1722         CFRunLoopRemoveSource((CFRunLoopRef)CFArrayGetValueAtIndex(runLoopsAndModes, i),
1723                               source,
1724                               (CFStringRef)CFArrayGetValueAtIndex(runLoopsAndModes, i + 1));
1725         }
1726 }
1727
1728 Boolean _CFStreamRemoveRunLoopAndModeFromArray(CFMutableArrayRef runLoopsAndModes, CFRunLoopRef rl, CFStringRef mode) {
1729     CFIndex idx, cnt;
1730     Boolean found = FALSE;
1731
1732     if (!runLoopsAndModes) return FALSE;
1733
1734     checkRLMArray(runLoopsAndModes);
1735
1736     cnt = CFArrayGetCount(runLoopsAndModes);
1737     for (idx = 0; idx + 1 < cnt; idx += 2) {
1738         if (CFEqual(CFArrayGetValueAtIndex(runLoopsAndModes, idx), rl) && CFEqual(CFArrayGetValueAtIndex(runLoopsAndModes, idx + 1), mode)) {
1739             CFArrayRemoveValueAtIndex(runLoopsAndModes, idx);
1740             CFArrayRemoveValueAtIndex(runLoopsAndModes, idx);
1741             found = TRUE;
1742             break;
1743         }
1744     }
1745     return found;
1746 }
1747
1748 // Used by NSStream to properly allocate the bridged objects
1749 CF_EXPORT CFIndex _CFStreamInstanceSize(void) {
1750     return sizeof(struct _CFStream);
1751 }
1752
1753 #if DEPLOYMENT_TARGET_MACOSX || DEPLOYMENT_TARGET_EMBEDDED
1754 #elif DEPLOYMENT_TARGET_WINDOWS
1755 void __CFStreamCleanup(void) {
1756     __CFSpinLock(&sSourceLock);
1757     if (sSharedSources) {
1758         CFIndex count = CFDictionaryGetCount(sSharedSources);
1759         if (count == 0) {
1760             // Only release if empty.  If it's still holding streams (which would be a client
1761             // bug leak), freeing this dict would free the streams, which then need to access the
1762             // dict to remove themselves, which leads to a deadlock.
1763             CFRelease(sSharedSources);
1764             sSharedSources = NULL;
1765         } else {
1766             const void ** keys = (const void **)malloc(sizeof(const void *) * count);
1767 #if defined(DEBUG)
1768             int i;
1769 #endif
1770             CFDictionaryGetKeysAndValues(sSharedSources, keys, NULL);
1771              fprintf(stderr, "*** CFNetwork is shutting down, but %ld streams are still scheduled.\n", count);
1772 #if defined(DEBUG)
1773             for (i = 0; i < count;i ++) {
1774                 if ((CFGetTypeID(keys[i]) == __kCFReadStreamTypeID) || (CFGetTypeID(keys[i]) == __kCFWriteStreamTypeID)) {
1775                     CFShow(keys[i]);
1776                 }
1777             }
1778 #endif
1779         }
1780     }
1781     __CFSpinUnlock(&sSourceLock);
1782 }
1783 #else
1784 #error Unknown or unspecified DEPLOYMENT_TARGET
1785 #endif
1786