Stream.subproj/CFConcreteStreams.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 /*      CFConcreteStreams.c
  24         Copyright 2000-2002, Apple, Inc. All rights reserved.
  25         Responsibility: Becky Willrich
  26 */
  27
  28 #include <sys/time.h>
  29 #include <stdlib.h>
  30 #include <unistd.h>
  31 #include <fcntl.h>
  32 #include <string.h>
  33 #include <stdio.h>
  34 #include <CoreFoundation/CFNumber.h>
  35 #include "CFStream.h"
  36 #include "CFStreamPriv.h"
  37 #include "CFInternal.h"
  38 #include "CFUtilitiesPriv.h"
  39
  40 // On Unix, you can schedule an fd with the RunLoop by creating a CFSocket around it.  On Win32
  41 // files and sockets are not interchangeable, and we do cheapo scheduling, where the file is
  42 // always readable and writable until we hit EOF (similar to the way CFData streams are scheduled).
  43 #if !defined(__WIN32__)
  44 #define REAL_FILE_SCHEDULING (1)
  45 #endif
  46
  47 #define SCHEDULE_AFTER_WRITE  (0)
  48 #define SCHEDULE_AFTER_READ   (1)
  49 #define APPEND                (3)
  50 #define AT_EOF                (4)
  51
  52 /* File callbacks */
  53 typedef struct {
  54     CFURLRef                            url;
  55     int                                         fd;
  56 #ifdef REAL_FILE_SCHEDULING
  57     union {
  58         CFSocketRef                     sock;           // socket created once we open and have an fd
  59         CFMutableArrayRef       rlArray;        // scheduling information prior to open
  60     } rlInfo; // If fd > 0, sock exists.  Otherwise, rlArray.
  61 #else
  62     uint16_t                            scheduled;      // ref count of how many times we've been scheduled
  63 #endif
  64     CFOptionFlags                       flags;
  65
  66     off_t                                       offset;
  67 } _CFFileStreamContext;
  68
  69
  70 CONST_STRING_DECL(kCFStreamPropertyFileCurrentOffset, "kCFStreamPropertyFileCurrentOffset");
  71
  72
  73 #ifdef REAL_FILE_SCHEDULING
  74 static void fileCallBack(CFSocketRef s, CFSocketCallBackType type, CFDataRef address, const void *data, void *info);
  75
  76 static void constructCFSocket(_CFFileStreamContext *fileStream, Boolean forRead, struct _CFStream *stream) {
  77     CFSocketContext context = {0, stream, NULL, NULL, CFCopyDescription};
  78     CFSocketRef sock = CFSocketCreateWithNative(CFGetAllocator(stream), fileStream->fd, forRead ? kCFSocketReadCallBack : kCFSocketWriteCallBack, fileCallBack, &context);
  79     CFSocketSetSocketFlags(sock, 0);
  80     if (fileStream->rlInfo.rlArray) {
  81         CFIndex i, c = CFArrayGetCount(fileStream->rlInfo.rlArray);
  82         CFRunLoopSourceRef src = CFSocketCreateRunLoopSource(CFGetAllocator(stream), sock, 0);
  83         for (i = 0; i+1 < c; i += 2) {
  84             CFRunLoopRef rl = (CFRunLoopRef)CFArrayGetValueAtIndex(fileStream->rlInfo.rlArray, i);
  85             CFStringRef mode = CFArrayGetValueAtIndex(fileStream->rlInfo.rlArray, i+1);
  86             CFRunLoopAddSource(rl, src, mode);
  87         }
  88         CFRelease(fileStream->rlInfo.rlArray);
  89         CFRelease(src);
  90     }
  91     fileStream->rlInfo.sock = sock;
  92 }
  93 #endif
  94
  95 static Boolean constructFD(_CFFileStreamContext *fileStream, CFStreamError *error, Boolean forRead, struct _CFStream *stream) {
  96     UInt8 path[1024];
  97     int flags = forRead ? O_RDONLY : (O_CREAT | O_TRUNC | O_WRONLY);
  98 #if defined(__WIN32__)
  99     flags |= (_O_BINARY|_O_NOINHERIT);
 100 #endif
 101
 102 __CFSetNastyFile(fileStream->url);
 103
 104     if (CFURLGetFileSystemRepresentation(fileStream->url, TRUE, path, 1024) == FALSE) {
 105         error->error = ENOENT;
 106         error->domain = kCFStreamErrorDomainPOSIX;
 107         return FALSE;
 108     }
 109     if (__CFBitIsSet(fileStream->flags, APPEND)) {
 110         flags |= O_APPEND;
 111         if(_CFExecutableLinkedOnOrAfter(CFSystemVersionPanther)) flags &= ~O_TRUNC;
 112     }
 113
 114     do {
 115         fileStream->fd = open(path, flags, 0666);
 116
 117         if (fileStream->fd < 0)
 118             break;
 119
 120         if ((fileStream->offset != -1) && (lseek(fileStream->fd, fileStream->offset, SEEK_SET) == -1))
 121             break;
 122
 123 #ifdef REAL_FILE_SCHEDULING
 124         if (fileStream->rlInfo.rlArray != NULL) {
 125             constructCFSocket(fileStream, forRead, stream);
 126         }
 127 #endif
 128
 129         return TRUE;
 130     } while (1);
 131
 132     error->error = errno;
 133     error->domain = kCFStreamErrorDomainPOSIX;
 134
 135     return FALSE;
 136 }
 137
 138 static Boolean fileOpen(struct _CFStream *stream, CFStreamError *errorCode, Boolean *openComplete, void *info) {
 139     _CFFileStreamContext *ctxt = (_CFFileStreamContext *)info;
 140     if (ctxt->fd >= 0) {
 141         // Open already occurred
 142         errorCode->error = 0;
 143         *openComplete = TRUE;
 144         return TRUE;
 145     }
 146     Boolean forRead = (CFGetTypeID(stream) == CFReadStreamGetTypeID());
 147     if (constructFD(ctxt, errorCode, forRead, stream)) {
 148         *openComplete = TRUE;
 149 #ifndef REAL_FILE_SCHEDULING
 150         if (ctxt->scheduled > 0) {
 151             if (forRead)
 152                 CFReadStreamSignalEvent((CFReadStreamRef)stream, kCFStreamEventHasBytesAvailable, NULL);
 153             else
 154                 CFWriteStreamSignalEvent((CFWriteStreamRef)stream, kCFStreamEventCanAcceptBytes, NULL);
 155         }
 156 #endif
 157         return TRUE;
 158     } else {
 159         return FALSE;
 160     }
 161 }
 162
 163 __private_extern__ CFIndex fdRead(int fd, UInt8 *buffer, CFIndex bufferLength, CFStreamError *errorCode, Boolean *atEOF) {
 164     CFIndex bytesRead = read(fd, buffer, bufferLength);
 165     if (bytesRead < 0) {
 166         errorCode->error = errno;
 167         errorCode->domain = kCFStreamErrorDomainPOSIX;
 168         return -1;
 169     } else {
 170         *atEOF = (bytesRead == 0) ? TRUE : FALSE;
 171         errorCode->error = 0;
 172         return bytesRead;
 173     }
 174 }
 175
 176 static CFIndex fileRead(CFReadStreamRef stream, UInt8 *buffer, CFIndex bufferLength, CFStreamError *errorCode, Boolean *atEOF, void *info) {
 177     _CFFileStreamContext *ctxt = (_CFFileStreamContext *)info;
 178     CFIndex result;
 179     result = fdRead(ctxt->fd, buffer, bufferLength, errorCode, atEOF);
 180 #ifdef REAL_FILE_SCHEDULING
 181     if (__CFBitIsSet(ctxt->flags, SCHEDULE_AFTER_READ)) {
 182         __CFBitClear(ctxt->flags, SCHEDULE_AFTER_READ);
 183         if (ctxt->rlInfo.sock) {
 184             CFSocketEnableCallBacks(ctxt->rlInfo.sock, kCFSocketReadCallBack);
 185         }
 186     }
 187 #else
 188     if (*atEOF)
 189         __CFBitSet(ctxt->flags, AT_EOF);
 190     if (ctxt->scheduled > 0 && !*atEOF) {
 191         CFReadStreamSignalEvent(stream, kCFStreamEventHasBytesAvailable, NULL);
 192     }
 193 #endif
 194     return result;
 195 }
 196
 197 #ifdef REAL_FILE_SCHEDULING
 198 __private_extern__ Boolean fdCanRead(int fd) {
 199     struct timeval timeout = {0, 0};
 200     fd_set *readSetPtr;
 201     fd_set readSet;
 202     Boolean result;
 203 // fd_set is not a mask in Win32, so checking for an fd that's too big is not relevant
 204     if (fd < FD_SETSIZE) {
 205         FD_ZERO(&readSet);
 206         readSetPtr = &readSet;
 207     } else {
 208         int size = howmany(fd+1, NFDBITS) * sizeof(uint32_t);
 209         uint32_t *fds_bits = (uint32_t *)malloc(size);
 210         memset(fds_bits, 0, size);
 211         readSetPtr = (fd_set *)fds_bits;
 212     }
 213     FD_SET(fd, readSetPtr);
 214     result = (select(fd + 1, readSetPtr, NULL, NULL, &timeout) == 1) ? TRUE : FALSE;
 215     if (readSetPtr != &readSet) {
 216         free(readSetPtr);
 217     }
 218     return result;
 219 }
 220 #endif
 221
 222 static Boolean fileCanRead(CFReadStreamRef stream, void *info) {
 223     _CFFileStreamContext *ctxt = (_CFFileStreamContext *)info;
 224 #ifdef REAL_FILE_SCHEDULING
 225     return fdCanRead(ctxt->fd);
 226 #else
 227     return !__CFBitIsSet(ctxt->flags, AT_EOF);
 228 #endif
 229 }
 230
 231 __private_extern__ CFIndex fdWrite(int fd, const UInt8 *buffer, CFIndex bufferLength, CFStreamError *errorCode) {
 232     CFIndex bytesWritten = write(fd, buffer, bufferLength);
 233     if (bytesWritten < 0) {
 234         errorCode->error = errno;
 235         errorCode->domain = kCFStreamErrorDomainPOSIX;
 236         return -1;
 237     } else {
 238         errorCode->error = 0;
 239         return bytesWritten;
 240     }
 241 }
 242
 243 static CFIndex fileWrite(CFWriteStreamRef stream, const UInt8 *buffer, CFIndex bufferLength, CFStreamError *errorCode, void *info) {
 244     _CFFileStreamContext *fileStream = ((_CFFileStreamContext *)info);
 245     CFIndex result = fdWrite(fileStream->fd, buffer, bufferLength, errorCode);
 246 #ifdef REAL_FILE_SCHEDULING
 247     if (__CFBitIsSet(fileStream->flags, SCHEDULE_AFTER_WRITE)) {
 248         __CFBitClear(fileStream->flags, SCHEDULE_AFTER_WRITE);
 249         if (fileStream->rlInfo.sock) {
 250             CFSocketEnableCallBacks(fileStream->rlInfo.sock, kCFSocketWriteCallBack);
 251         }
 252     }
 253 #else
 254     if (fileStream->scheduled > 0) {
 255         CFWriteStreamSignalEvent(stream, kCFStreamEventCanAcceptBytes, NULL);
 256     }
 257 #endif
 258     return result;
 259 }
 260
 261 #ifdef REAL_FILE_SCHEDULING
 262 __private_extern__ Boolean fdCanWrite(int fd) {
 263     struct timeval timeout = {0, 0};
 264     fd_set *writeSetPtr;
 265     fd_set writeSet;
 266     Boolean result;
 267     if (fd < FD_SETSIZE) {
 268         FD_ZERO(&writeSet);
 269         writeSetPtr = &writeSet;
 270     } else {
 271         int size = howmany(fd+1, NFDBITS) * sizeof(uint32_t);
 272         uint32_t *fds_bits = (uint32_t *)malloc(size);
 273         memset(fds_bits, 0, size);
 274         writeSetPtr = (fd_set *)fds_bits;
 275     }
 276     FD_SET(fd, writeSetPtr);
 277     result = (select(fd + 1, NULL, writeSetPtr, NULL, &timeout) == 1) ? TRUE : FALSE;
 278     if (writeSetPtr != &writeSet) {
 279         free(writeSetPtr);
 280     }
 281     return result;
 282 }
 283 #endif
 284
 285 static Boolean fileCanWrite(CFWriteStreamRef stream, void *info) {
 286 #ifdef REAL_FILE_SCHEDULING
 287     return fdCanWrite(((_CFFileStreamContext *)info)->fd);
 288 #else
 289     return TRUE;
 290 #endif
 291 }
 292
 293 static void fileClose(struct _CFStream *stream, void *info) {
 294     _CFFileStreamContext *ctxt = (_CFFileStreamContext *)info;
 295     if (ctxt->fd >= 0) {
 296         close(ctxt->fd);
 297         ctxt->fd = -1;
 298 #ifdef REAL_FILE_SCHEDULING
 299         if (ctxt->rlInfo.sock) {
 300             CFSocketInvalidate(ctxt->rlInfo.sock);
 301             CFRelease(ctxt->rlInfo.sock);
 302             ctxt->rlInfo.sock = NULL;
 303         }
 304     } else if (ctxt->rlInfo.rlArray) {
 305         CFRelease(ctxt->rlInfo.rlArray);
 306         ctxt->rlInfo.rlArray = NULL;
 307 #endif
 308     }
 309 }
 310
 311 #ifdef REAL_FILE_SCHEDULING
 312 static void fileCallBack(CFSocketRef s, CFSocketCallBackType type, CFDataRef address, const void *data, void *info) {
 313     struct _CFStream *stream = (struct _CFStream *)info;
 314     Boolean isReadStream = (CFGetTypeID(stream) == CFReadStreamGetTypeID());
 315     _CFFileStreamContext *fileStream = isReadStream ? CFReadStreamGetInfoPointer((CFReadStreamRef)stream) : CFWriteStreamGetInfoPointer((CFWriteStreamRef)stream);
 316     if (type == kCFSocketWriteCallBack) {
 317         __CFBitSet(fileStream->flags, SCHEDULE_AFTER_WRITE);
 318         CFWriteStreamSignalEvent((CFWriteStreamRef)stream, kCFStreamEventCanAcceptBytes, NULL);
 319     } else {
 320         // type == kCFSocketReadCallBack
 321         __CFBitSet(fileStream->flags, SCHEDULE_AFTER_READ);
 322         CFReadStreamSignalEvent((CFReadStreamRef)stream, kCFStreamEventHasBytesAvailable, NULL);
 323     }
 324 }
 325 #endif
 326
 327 static void fileSchedule(struct _CFStream *stream, CFRunLoopRef runLoop, CFStringRef runLoopMode, void *info) {
 328     _CFFileStreamContext *fileStream = (_CFFileStreamContext *)info;
 329     Boolean isReadStream = (CFGetTypeID(stream) == CFReadStreamGetTypeID());
 330     CFStreamStatus status = isReadStream ? CFReadStreamGetStatus((CFReadStreamRef)stream) : CFWriteStreamGetStatus((CFWriteStreamRef)stream);
 331     if (fileStream->fd < 0 && status != kCFStreamStatusNotOpen) {
 332         // Stream's already closed or error-ed out
 333         return;
 334     }
 335 #ifdef REAL_FILE_SCHEDULING
 336     if (fileStream->fd < 0) {
 337         if (!fileStream->rlInfo.rlArray) {
 338             fileStream->rlInfo.rlArray = CFArrayCreateMutable(CFGetAllocator(stream), 0, &kCFTypeArrayCallBacks);
 339         }
 340         CFArrayAppendValue(fileStream->rlInfo.rlArray, runLoop);
 341         CFArrayAppendValue(fileStream->rlInfo.rlArray, runLoopMode);
 342     } else {
 343         CFRunLoopSourceRef rlSrc;
 344         if (!fileStream->rlInfo.sock) {
 345             constructCFSocket(fileStream, isReadStream, stream);
 346         }
 347         rlSrc = CFSocketCreateRunLoopSource(CFGetAllocator(stream), fileStream->rlInfo.sock, 0);
 348         CFRunLoopAddSource(runLoop, rlSrc, runLoopMode);
 349         CFRelease(rlSrc);
 350     }
 351 #else
 352     fileStream->scheduled++;
 353     if (fileStream->scheduled == 1 && fileStream->fd > 0 && status == kCFStreamStatusOpen) {
 354         if (isReadStream)
 355             CFReadStreamSignalEvent((CFReadStreamRef)stream, kCFStreamEventHasBytesAvailable, NULL);
 356         else
 357             CFWriteStreamSignalEvent((CFWriteStreamRef)stream, kCFStreamEventCanAcceptBytes, NULL);
 358     }
 359 #endif
 360 }
 361
 362 static void fileUnschedule(struct _CFStream *stream, CFRunLoopRef runLoop, CFStringRef runLoopMode, void *info) {
 363     _CFFileStreamContext *fileStream = (_CFFileStreamContext *)info;
 364 #ifdef REAL_FILE_SCHEDULING
 365     if (fileStream->fd < 0) {
 366         // Not opened yet
 367         if (fileStream->rlInfo.rlArray) {
 368             CFMutableArrayRef runloops = fileStream->rlInfo.rlArray;
 369             CFIndex i, c;
 370             for (i = 0, c = CFArrayGetCount(runloops); i+1 < c; i += 2) {
 371                 if (CFEqual(CFArrayGetValueAtIndex(runloops, i), runLoop) && CFEqual(CFArrayGetValueAtIndex(runloops, i+1), runLoopMode)) {
 372                     CFArrayRemoveValueAtIndex(runloops, i);
 373                     CFArrayRemoveValueAtIndex(runloops, i);
 374                     break;
 375                 }
 376             }
 377         }
 378     } else if (fileStream->rlInfo.sock) {
 379         CFRunLoopSourceRef sockSource = CFSocketCreateRunLoopSource(CFGetAllocator(stream), fileStream->rlInfo.sock, 0);
 380         CFRunLoopRemoveSource(runLoop, sockSource, runLoopMode);
 381         CFRelease(sockSource);
 382     }
 383 #else
 384     if (fileStream->scheduled > 0)
 385         fileStream->scheduled--;
 386 #endif
 387 }
 388
 389 static CFTypeRef fileCopyProperty(struct _CFStream *stream, CFStringRef propertyName, void *info) {
 390
 391     CFTypeRef result = NULL;
 392     _CFFileStreamContext *fileStream = (_CFFileStreamContext *)info;
 393
 394     if (CFEqual(propertyName, kCFStreamPropertyFileCurrentOffset)) {
 395
 396         // NOTE that this does a lseek of 0 from the current location in
 397         // order to populate the offset field which will then be used to
 398         // create the resulting value.
 399         if (!__CFBitIsSet(fileStream->flags, APPEND) && fileStream->fd != -1) {
 400             fileStream->offset = lseek(fileStream->fd, 0, SEEK_CUR);
 401         }
 402
 403         if (fileStream->offset != -1) {
 404             result = CFNumberCreate(CFGetAllocator((CFTypeRef)stream), kCFNumberSInt64Type, &(fileStream->offset));
 405         }
 406     }
 407
 408     return result;
 409 }
 410
 411 static Boolean fileSetProperty(struct _CFStream *stream, CFStringRef prop, CFTypeRef val, void *info) {
 412
 413     Boolean result = FALSE;
 414     _CFFileStreamContext *fileStream = (_CFFileStreamContext *)info;
 415
 416     if (CFEqual(prop, kCFStreamPropertyAppendToFile) && CFGetTypeID(stream) == CFWriteStreamGetTypeID() &&
 417         CFWriteStreamGetStatus((CFWriteStreamRef)stream) == kCFStreamStatusNotOpen)
 418     {
 419         if (val == kCFBooleanTrue) {
 420             __CFBitSet(fileStream->flags, APPEND);
 421             fileStream->offset = -1;                            // Can't offset and append on the stream
 422         } else {
 423             __CFBitClear(fileStream->flags, APPEND);
 424         }
 425         result = TRUE;
 426     }
 427
 428     else if (CFEqual(prop, kCFStreamPropertyFileCurrentOffset)) {
 429
 430         if (!__CFBitIsSet(fileStream->flags, APPEND))
 431         {
 432             result = CFNumberGetValue((CFNumberRef)val, kCFNumberSInt64Type, &(fileStream->offset));
 433         }
 434
 435         if ((fileStream->fd != -1) && (lseek(fileStream->fd, fileStream->offset, SEEK_SET) == -1)) {
 436             result = FALSE;
 437         }
 438     }
 439
 440     return result;
 441 }
 442
 443 static void *fileCreate(struct _CFStream *stream, void *info) {
 444     _CFFileStreamContext *ctxt = (_CFFileStreamContext *)info;
 445     _CFFileStreamContext *newCtxt = CFAllocatorAllocate(CFGetAllocator(stream), sizeof(_CFFileStreamContext), 0);
 446     if (!newCtxt) return NULL;
 447     newCtxt->url = CFRetain(ctxt->url);
 448     newCtxt->fd = ctxt->fd;
 449 #ifdef REAL_FILE_SCHEDULING
 450     newCtxt->rlInfo.sock = NULL;
 451 #else
 452     newCtxt->scheduled = 0;
 453 #endif
 454     newCtxt->flags = 0;
 455     newCtxt->offset = -1;
 456     return newCtxt;
 457 }
 458
 459 static void     fileFinalize(struct _CFStream *stream, void *info) {
 460     _CFFileStreamContext *ctxt = (_CFFileStreamContext *)info;
 461     if (ctxt->fd > 0) {
 462 #ifdef REAL_FILE_SCHEDULING
 463         if (ctxt->rlInfo.sock) {
 464             CFSocketInvalidate(ctxt->rlInfo.sock);
 465             CFRelease(ctxt->rlInfo.sock);
 466         }
 467 #endif
 468         close(ctxt->fd);
 469 #ifdef REAL_FILE_SCHEDULING
 470     } else if (ctxt->rlInfo.rlArray) {
 471         CFRelease(ctxt->rlInfo.rlArray);
 472 #endif
 473     }
 474     CFRelease(ctxt->url);
 475     CFAllocatorDeallocate(CFGetAllocator(stream), ctxt);
 476 }
 477
 478 static CFStringRef fileCopyDescription(struct _CFStream *stream, void *info) {
 479     // This needs work
 480     return CFCopyDescription(((_CFFileStreamContext *)info)->url);
 481 }
 482
 483 /* CFData stream callbacks */
 484 typedef struct {
 485     CFDataRef data; // Mutable if the stream was constructed writable
 486     const UInt8 *loc; // Current location in the file
 487     Boolean scheduled;
 488     char _padding[3];
 489 } _CFReadDataStreamContext;
 490
 491 #define BUF_SIZE 1024
 492 typedef struct _CFStreamByteBuffer {
 493     UInt8 *bytes;
 494     CFIndex capacity, length;
 495     struct _CFStreamByteBuffer *next;
 496 } _CFStreamByteBuffer;
 497
 498 typedef struct {
 499     _CFStreamByteBuffer *firstBuf, *currentBuf;
 500     CFAllocatorRef bufferAllocator;
 501     Boolean scheduled;
 502     char _padding[3];
 503 } _CFWriteDataStreamContext;
 504
 505 static Boolean readDataOpen(struct _CFStream *stream, CFStreamError *errorCode, Boolean *openComplete, void *info) {
 506     _CFReadDataStreamContext *dataStream = (_CFReadDataStreamContext *)info;
 507     if (dataStream->scheduled) {
 508         if (CFDataGetLength(dataStream->data) != 0) {
 509             CFReadStreamSignalEvent((CFReadStreamRef)stream, kCFStreamEventHasBytesAvailable, NULL);
 510         } else {
 511             CFReadStreamSignalEvent((CFReadStreamRef)stream, kCFStreamEventEndEncountered, NULL);
 512         }
 513     }
 514     errorCode->error = 0;
 515     *openComplete = TRUE;
 516     return TRUE;
 517 }
 518
 519 static void readDataSchedule(struct _CFStream *stream, CFRunLoopRef rl, CFStringRef rlMode, void *info) {
 520     _CFReadDataStreamContext *dataStream = (_CFReadDataStreamContext *)info;
 521     if (dataStream->scheduled == FALSE) {
 522         dataStream->scheduled = TRUE;
 523                 if (CFReadStreamGetStatus((CFReadStreamRef)stream) != kCFStreamStatusOpen)
 524                         return;
 525         if (CFDataGetBytePtr(dataStream->data) + CFDataGetLength(dataStream->data) > dataStream->loc) {
 526             CFReadStreamSignalEvent((CFReadStreamRef)stream, kCFStreamEventHasBytesAvailable, NULL);
 527         } else {
 528             CFReadStreamSignalEvent((CFReadStreamRef)stream, kCFStreamEventEndEncountered, NULL);
 529         }
 530     }
 531 }
 532
 533 static CFIndex dataRead(CFReadStreamRef stream, UInt8 *buffer, CFIndex bufferLength, CFStreamError *error, Boolean *atEOF, void *info) {
 534     _CFReadDataStreamContext *dataCtxt = (_CFReadDataStreamContext *)info;
 535     const UInt8 *bytePtr = CFDataGetBytePtr(dataCtxt->data);
 536     CFIndex length = CFDataGetLength(dataCtxt->data);
 537     CFIndex bytesToCopy = bytePtr + length - dataCtxt->loc;
 538     if (bytesToCopy > bufferLength) {
 539         bytesToCopy = bufferLength;
 540     }
 541     if (bytesToCopy < 0) {
 542         bytesToCopy = 0;
 543     }
 544     if (bytesToCopy != 0) {
 545         memmove(buffer, dataCtxt->loc, bytesToCopy);
 546         dataCtxt->loc += bytesToCopy;
 547     }
 548     error->error = 0;
 549     *atEOF = (dataCtxt->loc < bytePtr + length) ? FALSE : TRUE;
 550     if (dataCtxt->scheduled && !*atEOF) {
 551         CFReadStreamSignalEvent(stream, kCFStreamEventHasBytesAvailable, NULL);
 552     }
 553     return bytesToCopy;
 554 }
 555
 556 static const UInt8 *dataGetBuffer(CFReadStreamRef stream, CFIndex maxBytesToRead, CFIndex *numBytesRead, CFStreamError *error, Boolean *atEOF, void *info) {
 557     _CFReadDataStreamContext *dataCtxt = (_CFReadDataStreamContext *)info;
 558     const UInt8 *bytes = CFDataGetBytePtr(dataCtxt->data);
 559     if (dataCtxt->loc - bytes > maxBytesToRead) {
 560         *numBytesRead = maxBytesToRead;
 561         *atEOF = FALSE;
 562     } else {
 563         *numBytesRead = dataCtxt->loc - bytes;
 564         *atEOF = TRUE;
 565     }
 566     error->error = 0;
 567     bytes = dataCtxt->loc;
 568     dataCtxt->loc += *numBytesRead;
 569     if (dataCtxt->scheduled && !*atEOF) {
 570         CFReadStreamSignalEvent(stream, kCFStreamEventHasBytesAvailable, NULL);
 571     }
 572     return bytes;
 573 }
 574
 575 static Boolean dataCanRead(CFReadStreamRef stream, void *info) {
 576     _CFReadDataStreamContext *dataCtxt = (_CFReadDataStreamContext *)info;
 577     return (CFDataGetBytePtr(dataCtxt->data) + CFDataGetLength(dataCtxt->data) > dataCtxt->loc) ? TRUE : FALSE;
 578 }
 579
 580 static Boolean writeDataOpen(struct _CFStream *stream, CFStreamError *errorCode, Boolean *openComplete, void *info) {
 581     _CFWriteDataStreamContext *dataStream = (_CFWriteDataStreamContext *)info;
 582     if (dataStream->scheduled) {
 583         if (dataStream->bufferAllocator != kCFAllocatorNull || dataStream->currentBuf->capacity > dataStream->currentBuf->length) {
 584             CFWriteStreamSignalEvent((CFWriteStreamRef)stream, kCFStreamEventCanAcceptBytes, NULL);
 585         } else {
 586             CFWriteStreamSignalEvent((CFWriteStreamRef)stream, kCFStreamEventEndEncountered, NULL);
 587         }
 588     }
 589     errorCode->error = 0;
 590     *openComplete = TRUE;
 591     return TRUE;
 592 }
 593
 594 static void writeDataSchedule(struct _CFStream *stream, CFRunLoopRef rl, CFStringRef rlMode, void *info) {
 595     _CFWriteDataStreamContext *dataStream = (_CFWriteDataStreamContext *)info;
 596     if (dataStream->scheduled == FALSE) {
 597         dataStream->scheduled = TRUE;
 598                 if (CFWriteStreamGetStatus((CFWriteStreamRef)stream) != kCFStreamStatusOpen)
 599                         return;
 600         if (dataStream->bufferAllocator != kCFAllocatorNull || dataStream->currentBuf->capacity > dataStream->currentBuf->length) {
 601             CFWriteStreamSignalEvent((CFWriteStreamRef)stream, kCFStreamEventCanAcceptBytes, NULL);
 602         } else {
 603             CFWriteStreamSignalEvent((CFWriteStreamRef)stream, kCFStreamEventEndEncountered, NULL);
 604         }
 605     }
 606 }
 607
 608 static CFIndex dataWrite(CFWriteStreamRef stream, const UInt8 *buffer, CFIndex bufferLength, CFStreamError *errorCode, void *info) {
 609     _CFWriteDataStreamContext *dataStream = (_CFWriteDataStreamContext *)info;
 610     CFIndex result;
 611     CFIndex freeSpace = dataStream->currentBuf->capacity - dataStream->currentBuf->length;
 612     if (dataStream->bufferAllocator == kCFAllocatorNull && bufferLength > freeSpace) {
 613         errorCode->error = ENOMEM;
 614         errorCode->domain = kCFStreamErrorDomainPOSIX;
 615         return -1;
 616     } else {
 617         result = bufferLength;
 618         while (bufferLength > 0) {
 619             CFIndex amountToCopy = (bufferLength > freeSpace) ? freeSpace : bufferLength;
 620             if (freeSpace > 0) {
 621                 memmove(dataStream->currentBuf->bytes + dataStream->currentBuf->length, buffer, amountToCopy);
 622                 buffer += amountToCopy;
 623                 bufferLength -= amountToCopy;
 624                 dataStream->currentBuf->length += amountToCopy;
 625             }
 626             if (bufferLength > 0) {
 627                 CFIndex bufSize = BUF_SIZE > bufferLength ? BUF_SIZE : bufferLength;
 628                 _CFStreamByteBuffer *newBuf = (_CFStreamByteBuffer *)CFAllocatorAllocate(dataStream->bufferAllocator, sizeof(_CFStreamByteBuffer) + bufSize, 0);
 629                 newBuf->bytes = (UInt8 *)(newBuf + 1);
 630                 newBuf->capacity = bufSize;
 631                 newBuf->length = 0;
 632                 newBuf->next = NULL;
 633                 dataStream->currentBuf->next = newBuf;
 634                 dataStream->currentBuf = newBuf;
 635                 freeSpace = bufSize;
 636             }
 637         }
 638         errorCode->error = 0;
 639     }
 640     if (dataStream->scheduled && (dataStream->bufferAllocator != kCFAllocatorNull || dataStream->currentBuf->capacity > dataStream->currentBuf->length)) {
 641         CFWriteStreamSignalEvent(stream, kCFStreamEventCanAcceptBytes, NULL);
 642     }
 643     return result;
 644 }
 645
 646 static  Boolean dataCanWrite(CFWriteStreamRef stream, void *info) {
 647     _CFWriteDataStreamContext *dataStream = (_CFWriteDataStreamContext *)info;
 648     if (dataStream->bufferAllocator != kCFAllocatorNull) return TRUE;
 649     if (dataStream->currentBuf->capacity  > dataStream->currentBuf->length) return TRUE;
 650     return FALSE;
 651 }
 652
 653 static CFPropertyListRef dataCopyProperty(struct _CFStream *stream, CFStringRef propertyName, void *info) {
 654     _CFWriteDataStreamContext *dataStream = (_CFWriteDataStreamContext *)info;
 655     CFIndex size = 0;
 656     _CFStreamByteBuffer *buf;
 657     CFAllocatorRef alloc;
 658     UInt8 *bytes, *currByte;
 659     if (!CFEqual(propertyName, kCFStreamPropertyDataWritten)) return NULL;
 660     if (dataStream->bufferAllocator == kCFAllocatorNull)  return NULL;
 661     alloc = dataStream->bufferAllocator;
 662     for (buf = dataStream->firstBuf; buf != NULL; buf = buf->next) {
 663         size += buf->length;
 664     }
 665     if (size == 0) return NULL;
 666     bytes = CFAllocatorAllocate(alloc, size, 0);
 667     currByte = bytes;
 668     for (buf = dataStream->firstBuf; buf != NULL; buf = buf->next) {
 669         memmove(currByte, buf->bytes, buf->length);
 670         currByte += buf->length;
 671     }
 672     return CFDataCreateWithBytesNoCopy(alloc, bytes, size, alloc);
 673 }
 674
 675 static void *readDataCreate(struct _CFStream *stream, void *info) {
 676     _CFReadDataStreamContext *ctxt = (_CFReadDataStreamContext *)info;
 677     _CFReadDataStreamContext *newCtxt = (_CFReadDataStreamContext *)CFAllocatorAllocate(CFGetAllocator(stream), sizeof(_CFReadDataStreamContext), 0);
 678     if (!newCtxt) return NULL;
 679     newCtxt->data = CFRetain(ctxt->data);
 680     newCtxt->loc = CFDataGetBytePtr(newCtxt->data);
 681     newCtxt->scheduled = FALSE;
 682     return (void *)newCtxt;
 683 }
 684
 685 static void readDataFinalize(struct _CFStream *stream, void *info) {
 686     _CFReadDataStreamContext *ctxt = (_CFReadDataStreamContext *)info;
 687     CFRelease(ctxt->data);
 688     CFAllocatorDeallocate(CFGetAllocator(stream), ctxt);
 689 }
 690
 691 static CFStringRef readDataCopyDescription(struct _CFStream *stream, void *info) {
 692     return CFCopyDescription(((_CFReadDataStreamContext *)info)->data);
 693 }
 694
 695 static void *writeDataCreate(struct _CFStream *stream, void *info) {
 696     _CFWriteDataStreamContext *ctxt = (_CFWriteDataStreamContext *)info;
 697     _CFWriteDataStreamContext *newCtxt;
 698     if (ctxt->bufferAllocator != kCFAllocatorNull) {
 699         if (ctxt->bufferAllocator == NULL) ctxt->bufferAllocator = CFAllocatorGetDefault();
 700         CFRetain(ctxt->bufferAllocator);
 701         newCtxt = CFAllocatorAllocate(CFGetAllocator(stream), sizeof(_CFWriteDataStreamContext) + sizeof(_CFStreamByteBuffer) + BUF_SIZE, 0);
 702         newCtxt->firstBuf = (_CFStreamByteBuffer *)(newCtxt + 1);
 703         newCtxt->firstBuf->bytes = (UInt8 *)(newCtxt->firstBuf + 1);
 704         newCtxt->firstBuf->capacity = BUF_SIZE;
 705         newCtxt->firstBuf->length = 0;
 706         newCtxt->firstBuf->next = NULL;
 707         newCtxt->currentBuf = newCtxt->firstBuf;
 708         newCtxt->bufferAllocator = ctxt->bufferAllocator;
 709         newCtxt->scheduled = FALSE;
 710     } else {
 711         newCtxt = (_CFWriteDataStreamContext *)CFAllocatorAllocate(CFGetAllocator(stream), sizeof(_CFWriteDataStreamContext) + sizeof(_CFStreamByteBuffer), 0);
 712         newCtxt->firstBuf = (_CFStreamByteBuffer *)(newCtxt+1);
 713         newCtxt->firstBuf->bytes = ctxt->firstBuf->bytes;
 714         newCtxt->firstBuf->capacity = ctxt->firstBuf->capacity;
 715         newCtxt->firstBuf->length = 0;
 716         newCtxt->firstBuf->next = NULL;
 717         newCtxt->currentBuf = newCtxt->firstBuf;
 718         newCtxt->bufferAllocator = kCFAllocatorNull;
 719         newCtxt->scheduled = FALSE;
 720     }
 721     return (void *)newCtxt;
 722 }
 723
 724 static void writeDataFinalize(struct _CFStream *stream, void *info) {
 725     _CFWriteDataStreamContext *ctxt = (_CFWriteDataStreamContext *)info;
 726     if (ctxt->bufferAllocator != kCFAllocatorNull) {
 727         _CFStreamByteBuffer *buf = ctxt->firstBuf->next, *next;
 728         while (buf != NULL) {
 729             next = buf->next;
 730             CFAllocatorDeallocate(ctxt->bufferAllocator, buf);
 731             buf = next;
 732         }
 733         CFRelease(ctxt->bufferAllocator);
 734     }
 735     CFAllocatorDeallocate(CFGetAllocator(stream), ctxt);
 736 }
 737
 738 static CFStringRef writeDataCopyDescription(struct _CFStream *stream, void *info) {
 739     return CFStringCreateWithFormat(NULL, NULL, CFSTR("<CFWriteDataContext 0x%x>"), (int)info);
 740 }
 741
 742 static const struct _CFStreamCallBacks fileCallBacks = {1, fileCreate, fileFinalize, fileCopyDescription, fileOpen, NULL, fileRead, NULL, fileCanRead, fileWrite, fileCanWrite, fileClose, fileCopyProperty, fileSetProperty, NULL, fileSchedule, fileUnschedule};
 743
 744 static struct _CFStream *_CFStreamCreateWithFile(CFAllocatorRef alloc, CFURLRef fileURL, Boolean forReading) {
 745     _CFFileStreamContext fileContext;
 746     CFStringRef scheme = fileURL ? CFURLCopyScheme(fileURL) : NULL;
 747     if (!scheme || !CFEqual(scheme, CFSTR("file"))) {
 748         if (scheme) CFRelease(scheme);
 749         return NULL;
 750     }
 751     CFRelease(scheme);
 752     fileContext.url = fileURL;
 753     fileContext.fd = -1;
 754     return _CFStreamCreateWithConstantCallbacks(alloc, &fileContext, &fileCallBacks, forReading);
 755 }
 756
 757 CF_EXPORT CFReadStreamRef CFReadStreamCreateWithFile(CFAllocatorRef alloc, CFURLRef fileURL) {
 758     return (CFReadStreamRef)_CFStreamCreateWithFile(alloc, fileURL, TRUE);
 759 }
 760
 761 CF_EXPORT CFWriteStreamRef CFWriteStreamCreateWithFile(CFAllocatorRef alloc, CFURLRef fileURL) {
 762     return (CFWriteStreamRef)_CFStreamCreateWithFile(alloc, fileURL, FALSE);
 763 }
 764
 765 static const struct _CFStreamCallBacks readDataCallBacks = {1, readDataCreate, readDataFinalize, readDataCopyDescription, readDataOpen, NULL, dataRead, dataGetBuffer, dataCanRead, NULL, NULL, NULL, NULL, NULL, NULL, readDataSchedule, NULL};
 766 static const struct _CFStreamCallBacks writeDataCallBacks = {1, writeDataCreate, writeDataFinalize, writeDataCopyDescription, writeDataOpen, NULL, NULL, NULL, NULL, dataWrite, dataCanWrite, NULL, dataCopyProperty, NULL, NULL, writeDataSchedule, NULL};
 767
 768 CF_EXPORT CFReadStreamRef CFReadStreamCreateWithBytesNoCopy(CFAllocatorRef alloc, const UInt8 *bytes, CFIndex length, CFAllocatorRef bytesDeallocator) {
 769     _CFReadDataStreamContext ctxt;
 770     CFReadStreamRef result;
 771     ctxt.data = CFDataCreateWithBytesNoCopy(alloc, bytes, length, bytesDeallocator);
 772     result = (CFReadStreamRef)_CFStreamCreateWithConstantCallbacks(alloc, &ctxt, &readDataCallBacks, TRUE);
 773     CFRelease(ctxt.data);
 774     return result;
 775 }
 776
 777 CFWriteStreamRef CFWriteStreamCreateWithBuffer(CFAllocatorRef alloc, UInt8 *buffer, CFIndex bufferCapacity) {
 778     _CFStreamByteBuffer buf;
 779     _CFWriteDataStreamContext ctxt;
 780     buf.bytes = buffer;
 781     buf.capacity = bufferCapacity;
 782     buf.length = 0;
 783     buf.next = NULL;
 784     ctxt.firstBuf = &buf;
 785     ctxt.currentBuf = ctxt.firstBuf;
 786     ctxt.bufferAllocator = kCFAllocatorNull;
 787     return (CFWriteStreamRef)_CFStreamCreateWithConstantCallbacks(alloc, &ctxt, &writeDataCallBacks, FALSE);
 788 }
 789
 790 CF_EXPORT CFWriteStreamRef CFWriteStreamCreateWithAllocatedBuffers(CFAllocatorRef alloc, CFAllocatorRef bufferAllocator) {
 791     _CFWriteDataStreamContext ctxt;
 792     ctxt.firstBuf = NULL;
 793     ctxt.currentBuf = NULL;
 794     ctxt.bufferAllocator = bufferAllocator;
 795     return (CFWriteStreamRef)_CFStreamCreateWithConstantCallbacks(alloc, &ctxt, &writeDataCallBacks, FALSE);
 796 }
 797
 798 #undef BUF_SIZE