Stream.subproj/CFStream.c

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