#15 file stream reader IOS implementation

src/index.js

@@ -10,47 +10,19 @@ import {
   NativeAppEventEmitter,
   Platform,
 } from 'react-native'
-
-
-type fetchConfig = {
-  fileCache : bool,
-  path : string,
-  appendExt : string
-};
-
-type RNFetchBlobNative = {
-  fetchBlob : (
-    options:fetchConfig,
-    taskId:string,
-    method:string,
-    url:string,
-    headers:any,
-    body:any,
-    callback:(err:any, ...data:any) => void
-  ) => void,
-  fetchBlobForm : (
-    options:fetchConfig,
-    taskId:string,
-    method:string,
-    url:string,
-    headers:any,
-    form:Array<any>,
-    callback:(err:any, ...data:any) => void
-  ) => void,
-  readStream : (
-    path:string,
-    encode: 'utf8' | 'ascii' | 'base64'
-  ) => void,
-  getEnvironmentDirs : (dirs:any) => void,
-  flush : () => void
-};
-
+import type {
+  RNFetchBlobNative,
+  RNFetchBlobConfig,
+  RNFetchBlobStream
+}from './types'
 import base64 from 'base-64'
-const emitter = (Platform.OS === 'android' ? DeviceEventEmitter : NativeAppEventEmitter)
+
+// const emitter = (Platform.OS === 'android' ? DeviceEventEmitter : NativeAppEventEmitter)
+const emitter = DeviceEventEmitter
 const RNFetchBlob:RNFetchBlobNative = NativeModules.RNFetchBlob
 
 emitter.addListener("RNFetchBlobMessage", (e) => {
-
+  console.log(e)
   if(e.event === 'warn') {
     console.warn(e.detail)
   }
@@ -86,17 +58,16 @@ function getSystemDirs() {
 
 }
 
-function config (options:fetchConfig) {
+function config (options:RNFetchBlobConfig) {
   return { fetch : fetch.bind(options) }
 }
 
 // Promise wrapper function
 function fetch(...args:any) {
 
-  let options = this || {}
-
   // create task ID for receiving progress event
   let taskId = getUUID()
+  let options = this || {}
 
   let promise = new Promise((resolve, reject) => {
 
@@ -123,7 +94,6 @@ function fetch(...args:any) {
           respType = 'path'
         resolve(new FetchBlobResponse(taskId, respType, ...data))
       }
-
     })
 
   })
@@ -137,6 +107,47 @@ function fetch(...args:any) {
 
 }
 
+function openReadStream(path:string, encoding:'utf8' | 'ascii' | 'base64'):RNFetchBlobStream {
+
+  if(!path)
+    throw Error('RNFetchBlob could not open file stream with empty `path`')
+
+  let stream:RNFetchBlobStream = {
+    onData : function(fn) {
+      this._onData = fn
+    },
+    onError : function(fn) {
+      this._onError = fn
+    },
+    onEnd : function(fn) {
+      this._onEnd = fn
+    },
+  }
+
+  // register for file stream event
+  let subscription = emitter.addListener(`RNFetchBlobStream+${path}`, (e) => {
+    let {event, detail} = e
+    if(stream._onData && event === 'data')
+      stream._onData(detail)
+    else if (stream._onEnd && event === 'end') {
+      stream._onEnd(detail)
+    }
+    else {
+      stream._onError(detail)
+    }
+    // when stream closed or error, remove event handler
+    if (event === 'error' || event === 'end') {
+      subscription.remove()
+    }
+
+  })
+
+  RNFetchBlob.readStream(path, encoding)
+
+  return stream
+
+}
+
 /**
  * RNFetchBlob response object class.
  */
@@ -153,8 +164,7 @@ class FetchBlobResponse {
   flush : () => void;
   readStream : (
     encode: 'utf8' | 'ascii' | 'base64',
-    fn:(event : 'data' | 'end', chunk:any) => void
-  ) => void;
+  ) => RNFetchBlobStream | null;
 
   constructor(taskId:string, type:'base64' | 'path', data:any) {
     this.data = data
@@ -198,43 +208,36 @@ class FetchBlobResponse {
     this.flush = () => {
       RNFetchBlob.flush(this.path())
     }
-
+    /**
+     * get path of response temp file
+     * @return {string} File path of temp file.
+     */
     this.path = () => {
       if(this.type === 'path')
         return this.data
       return null
     }
-
     /**
      * Start read stream from cached file
      * @param  {String} encoding Encode type, should be one of `base64`, `ascrii`, `utf8`.
      * @param  {Function} fn On data event handler
      * @return {void}
      */
-    this.readStream = (encode: 'base64' | 'utf8' | 'ascii', fn) => {
-
-      // register for file stream event
-      let subscription = emitter.addListener(`RNFetchBlobStream${this.taskId}`, (event, chunk) => {
-        fn(event, chunk)
-        // when stream closed, remove event handler
-        if(event === 'end')
-          subscription()
-      })
-
+    this.readStream = (encode: 'base64' | 'utf8' | 'ascii'):RNFetchBlobStream | null => {
       if(this.type === 'path') {
-        RNFetchBlob.readStream(this.data, encode)
+        return openReadStream(this.data, encode)
       }
       else {
         console.warn('RNFetchblob', 'this response data does not contains any available stream')
+        return null
       }
-
     }
 
   }
 
 }
 
-function getUUID(){
+function getUUID() {
   return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function(c) {
     let r = Math.random()*16|0, v = c == 'x' ? r : (r&0x3|0x8);
     return v.toString(16);
@@ -242,5 +245,5 @@ function getUUID(){
 }
 
 export default {
-  fetch, base64, config, getSystemDirs
+  fetch, base64, config, getSystemDirs, openReadStream
 }

src/ios/RNFetchBlob/RNFetchBlob.h

@@ -46,6 +46,7 @@ extern NSString *const FS_EVENT_ERROR;
 @property (nonatomic) NSString * path;
 
 + (NSString *) getTempPath;
+- (id) init;
 - (void) initWithCallback;
 - (void) initWithBridgeRef;
 - (void) openWithDestination;
@@ -78,7 +79,6 @@ extern NSString *const FS_EVENT_ERROR;
 @property (nonatomic) FetchBlobFS * fileStream;
 
 - (id) init;
-- (id) delegate;
 - (void) sendRequest;
 
 + (NSMutableDictionary *) normalizeHeaders;

src/ios/RNFetchBlob/RNFetchBlob.m

@@ -42,13 +42,12 @@ NSString *const FS_EVENT_ERROR = @"error";
 
 
 
-+ (NSString *) getCacheDir {
 
++ (NSString *) getCacheDir {
     return [NSSearchPathForDirectoriesInDomains(NSCachesDirectory, NSUserDomainMask, YES) firstObject];
 }
 
 + (NSString *) getDocumentDir {
-
     return [NSSearchPathForDirectoriesInDomains(NSDocumentDirectory, NSUserDomainMask, YES) firstObject];
 }
 
@@ -75,6 +74,12 @@ NSString *const FS_EVENT_ERROR = @"error";
     return tempPath;
 }
 
+- (id)init {
+    self = [super init];
+    return self;
+}
+
+
 - (id)initWithCallback:(RCTResponseSenderBlock)callback {
     self = [super init];
     self.callback = callback;
@@ -83,12 +88,13 @@ NSString *const FS_EVENT_ERROR = @"error";
 
 - (id)initWithBridgeRef:(RCTBridge *)bridgeRef {
     self = [super init];
-    self.callback = callback;
+    self.bridge = bridgeRef;
     return self;
 }
 
 - (void)openWithPath:(NSString *)destPath {
     self.outStream = [[NSOutputStream alloc] initToFileAtPath:destPath append:YES];
+    [self.outStream scheduleInRunLoop:[NSRunLoop currentRunLoop] forMode:NSRunLoopCommonModes];
     [self.outStream open];
 }
 
@@ -108,13 +114,25 @@ NSString *const FS_EVENT_ERROR = @"error";
 }
 
 - (void)readWithPath:(NSString *)path useEncoding:(NSString *)encoding {
-    self.inStream = [[NSInputStream alloc]init];
+    
+    self.inStream = [[NSInputStream alloc] initWithFileAtPath:path];
+    self.inStream.delegate = self;
     self.encoding = encoding;
-    [self.inStream setDelegate:self];
+    self.path = path;
+    
+    // NSStream needs a runloop so let's create a run loop for it
+    dispatch_queue_t queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT,0);
+    // start NSStream is a runloop
+    dispatch_async(queue, ^ {
+        [inStream scheduleInRunLoop:[NSRunLoop currentRunLoop]
+                             forMode:NSDefaultRunLoopMode];
+        [inStream open];
+        [[NSRunLoop currentRunLoop] run];
+    
+    });
     
 }
 
-
 // close file write stream
 - (void)closeOutStream {
     if(self.outStream != nil) {
@@ -133,10 +151,25 @@ NSString *const FS_EVENT_ERROR = @"error";
     
 }
 
+void runOnMainQueueWithoutDeadlocking(void (^block)(void))
+{
+    if ([NSThread isMainThread])
+    {
+        block();
+    }
+    else
+    {
+        dispatch_sync(dispatch_get_main_queue(), block);
+    }
+}
+
+
 #pragma mark RNFetchBlobFS read stream delegate
 
 - (void)stream:(NSStream *)stream handleEvent:(NSStreamEvent)eventCode {
 
+    NSString * streamEventCode = [NSString stringWithFormat:@"RNFetchBlobStream+%@", self.path];
+    
     switch(eventCode) {
             
         // write stream event
@@ -149,10 +182,7 @@ NSString *const FS_EVENT_ERROR = @"error";
         // read stream incoming chunk
         case NSStreamEventHasBytesAvailable:
         {
-            
-            
             NSMutableData * chunkData = [[NSMutableData data] init];
-            
             uint8_t buf[1024];
             unsigned int len = 0;
             len = [(NSInputStream *)stream read:buf maxLength:1024];
@@ -160,41 +190,41 @@ NSString *const FS_EVENT_ERROR = @"error";
             if(len) {
                 [chunkData appendBytes:(const void *)buf length:len];
                 // TODO : file read progress ?
-//                [bytesRead setIntValue:[bytesRead intValue]+len];
-                
                 // dispatch data event
-                NSString * encodedChunk = [NSString alloc];
-                if( [self.encoding caseInsensitiveCompare:@"utf8"] ) {
+                NSString * encodedChunk;
+                if( [[self.encoding lowercaseString] isEqualToString:@"utf8"] ) {
                     encodedChunk = [encodedChunk initWithData:chunkData encoding:NSUTF8StringEncoding];
                 }
-                else if ( [self.encoding caseInsensitiveCompare:@"ascii"] ) {
+                else if ( [[self.encoding lowercaseString] isEqualToString:@"ascii"] ) {
                     encodedChunk = [encodedChunk initWithData:chunkData encoding:NSASCIIStringEncoding];
                 }
-                else if ( [self.encoding caseInsensitiveCompare:@"base64"] ) {
+                else if ( [[self.encoding lowercaseString] isEqualToString:@"base64"] ) {
                     encodedChunk = [chunkData base64EncodedStringWithOptions:0];
                 }
                 else {
                     [self.bridge.eventDispatcher
-                     sendAppEventWithName: [NSString stringWithFormat:@"RNFetchBlobStream%s", self.taskId]
-                     body:@{
+                        sendDeviceEventWithName:streamEventCode
+                        body:@{
                             @"event": FS_EVENT_ERROR,
                             @"detail": @"unrecognized encoding"
                         }
                      ];
                     return;
                 }
-                [self.bridge.eventDispatcher
-                 sendAppEventWithName: [NSString stringWithFormat:@"RNFetchBlobStream%s", self.taskId]
-                 body:@{
-                        @"event": FS_EVENT_DATA,
-                        @"detail": encodedChunk
-                    }
-                 ];
+                runOnMainQueueWithoutDeadlocking(^{
+                    [self.bridge.eventDispatcher
+                     sendDeviceEventWithName:streamEventCode
+                     body:@{
+                            @"event": FS_EVENT_DATA,
+                            @"detail": encodedChunk
+                            }
+                     ];
+                });
             }
             // end of stream
             else {
                 [self.bridge.eventDispatcher
-                sendAppEventWithName: [NSString stringWithFormat:@"RNFetchBlobStream%s", self.taskId]
+                sendDeviceEventWithName:streamEventCode
                 body:@{
                        @"event": FS_EVENT_END,
                        @"detail": @""
@@ -208,7 +238,7 @@ NSString *const FS_EVENT_ERROR = @"error";
         case NSStreamEventErrorOccurred:
         {
             [self.bridge.eventDispatcher
-             sendAppEventWithName: [NSString stringWithFormat:@"RNFetchBlobStream%s", self.taskId]
+             sendDeviceEventWithName:streamEventCode
              body:@{
                     @"event": FS_EVENT_ERROR,
                     @"detail": @"error when read file with stream"
@@ -318,7 +348,7 @@ NSString *const FS_EVENT_ERROR = @"error";
     }
     
     [self.bridge.eventDispatcher
-        sendAppEventWithName:@"RNFetchBlobProgress"
+        sendDeviceEventWithName:@"RNFetchBlobProgress"
         body:@{
             @"taskId": taskId,
             @"written": [NSString stringWithFormat:@"%d", receivedBytes],
@@ -332,7 +362,7 @@ NSString *const FS_EVENT_ERROR = @"error";
     expectedBytes = totalBytesExpectedToWrite;
     receivedBytes += totalBytesWritten;
     [self.bridge.eventDispatcher
-        sendAppEventWithName:@"RNFetchBlobProgress"
+        sendDeviceEventWithName:@"RNFetchBlobProgress"
             body:@{
                     @"taskId": taskId,
                     @"written": [NSString stringWithFormat:@"%d", receivedBytes],

src/types.js

@@ -0,0 +1,47 @@
+
+type RNFetchBlobConfig = {
+  fileCache : bool,
+  path : string,
+  appendExt : string
+};
+
+type RNFetchBlobNative = {
+  // API for fetch octet-stream data
+  fetchBlob : (
+    options:fetchConfig,
+    taskId:string,
+    method:string,
+    url:string,
+    headers:any,
+    body:any,
+    callback:(err:any, ...data:any) => void
+  ) => void,
+  // API for fetch form data
+  fetchBlobForm : (
+    options:fetchConfig,
+    taskId:string,
+    method:string,
+    url:string,
+    headers:any,
+    form:Array<any>,
+    callback:(err:any, ...data:any) => void
+  ) => void,
+  // open file stream
+  readStream : (
+    path:string,
+    encode: 'utf8' | 'ascii' | 'base64'
+  ) => void,
+  // get system folders
+  getEnvironmentDirs : (dirs:any) => void,
+  // unlink file by path
+  flush : (path:string) => void
+};
+
+type RNFetchBlobStream = {
+  onData : () => void,
+  onError : () => void,
+  onEnd : () => void,
+  _onData : () => void,
+  _onEnd : () => void,
+  _onError : () => void,
+}