CFStream.c

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