CFStream.c

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