//
// MQTTDecoder.m
// MQTTClient.framework
//
// Copyright © 2013-2017, Christoph Krey. All rights reserved.
//

#import "MQTTDecoder.h"

#import "MQTTLog.h"

@interface MQTTDecoder() {
    void *QueueIdentityKey;
}

@property (nonatomic) NSMutableArray<NSInputStream *> *streams;

@end

@implementation MQTTDecoder

- (instancetype)init {
    self = [super init];
    self.state = MQTTDecoderStateInitializing;
    self.streams = [NSMutableArray arrayWithCapacity:5];
    self.queue = dispatch_get_main_queue();
    return self;
}

- (void)dealloc {
    [self close];
}

- (void)setQueue:(dispatch_queue_t)queue {
    _queue = queue;
    
    // We're going to use dispatch_queue_set_specific() to "mark" our queue.
    // The dispatch_queue_set_specific() and dispatch_get_specific() functions take a "void *key" parameter.
    // Later we can use dispatch_get_specific() to determine if we're executing on our queue.
    // From the documentation:
    //
    // > Keys are only compared as pointers and are never dereferenced.
    // > Thus, you can use a pointer to a static variable for a specific subsystem or
    // > any other value that allows you to identify the value uniquely.
    //
    // So we're just going to use the memory address of an ivar.
    
    dispatch_queue_set_specific(_queue, &QueueIdentityKey, (__bridge void *)_queue, NULL);
}

- (void)decodeMessage:(NSData *)data {
    NSInputStream *stream = [NSInputStream inputStreamWithData:data];
    CFReadStreamRef readStream = (__bridge CFReadStreamRef)stream;
    CFReadStreamSetDispatchQueue(readStream, self.queue);
    [self openStream:stream];
}

- (void)openStream:(NSInputStream *)stream {
    [self.streams addObject:stream];
    stream.delegate = self;
    DDLogVerbose(@"[MQTTDecoder] #streams=%lu", (unsigned long)self.streams.count);
    if (self.streams.count == 1) {
        [stream open];
    }
}

- (void)open {
    self.state = MQTTDecoderStateDecodingHeader;
}

- (void)internalClose {
    if (self.streams) {
        for (NSInputStream *stream in self.streams) {
            [stream close];
            [stream setDelegate:nil];
        }
        [self.streams removeAllObjects];
    }
}

- (void)close {
    // https://github.com/novastone-media/MQTT-Client-Framework/issues/325
    // We need to make sure that we are closing streams on their queue
    // Otherwise, we end up with race condition where delegate is deallocated
    // but still used by run loop event
    if (self.queue != dispatch_get_specific(&QueueIdentityKey)) {
        dispatch_sync(self.queue, ^{
            [self internalClose];
        });
    } else {
        [self internalClose];
    }
}

- (void)stream:(NSStream *)sender handleEvent:(NSStreamEvent)eventCode {
    // We contact our delegate, MQTTSession at some point in this method
    // This call can cause MQTTSession to dealloc and thus, MQTTDecoder to dealloc
    // So we end up with invalid object in the middle of the method
    // To prevent this we retain self for duration of this method call
    MQTTDecoder *strongDecoder = self;
    (void)strongDecoder;
    
    NSInputStream *stream = (NSInputStream *)sender;
    
    if (eventCode & NSStreamEventOpenCompleted) {
        DDLogVerbose(@"[MQTTDecoder] NSStreamEventOpenCompleted");
    }
    
    if (eventCode & NSStreamEventHasBytesAvailable) {
        DDLogVerbose(@"[MQTTDecoder] NSStreamEventHasBytesAvailable");
        
        if (self.state == MQTTDecoderStateDecodingHeader) {
            UInt8 buffer;
            NSInteger n = [stream read:&buffer maxLength:1];
            if (n == -1) {
                self.state = MQTTDecoderStateConnectionError;
                [self.delegate decoder:self handleEvent:MQTTDecoderEventConnectionError error:stream.streamError];
            } else if (n == 1) {
                self.length = 0;
                self.lengthMultiplier = 1;
                self.state = MQTTDecoderStateDecodingLength;
                self.dataBuffer = [[NSMutableData alloc] init];
                [self.dataBuffer appendBytes:&buffer length:1];
                self.offset = 1;
                DDLogVerbose(@"[MQTTDecoder] fixedHeader=0x%02x", buffer);
            }
        }
        while (self.state == MQTTDecoderStateDecodingLength) {
            // TODO: check max packet length(prevent evil server response)
            UInt8 digit;
            NSInteger n = [stream read:&digit maxLength:1];
            if (n == -1) {
                self.state = MQTTDecoderStateConnectionError;
                [self.delegate decoder:self handleEvent:MQTTDecoderEventConnectionError error:stream.streamError];
                break;
            } else if (n == 0) {
                break;
            }
            DDLogVerbose(@"[MQTTDecoder] digit=0x%02x 0x%02x %d %d", digit, digit & 0x7f, (unsigned int)self.length, (unsigned int)self.lengthMultiplier);
            [self.dataBuffer appendBytes:&digit length:1];
            self.offset++;
            self.length += ((digit & 0x7f) * self.lengthMultiplier);
            if ((digit & 0x80) == 0x00) {
                self.state = MQTTDecoderStateDecodingData;
            } else {
                self.lengthMultiplier *= 128;
            }
        }
        DDLogVerbose(@"[MQTTDecoder] remainingLength=%d", (unsigned int)self.length);

        if (self.state == MQTTDecoderStateDecodingData) {
            if (self.length > 0) {
                NSInteger n, toRead;
                UInt8 buffer[768];
                toRead = self.length + self.offset - self.dataBuffer.length;
                if (toRead > sizeof buffer) {
                    toRead = sizeof buffer;
                }
                n = [stream read:buffer maxLength:toRead];
                if (n == -1) {
                    self.state = MQTTDecoderStateConnectionError;
                    [self.delegate decoder:self handleEvent:MQTTDecoderEventConnectionError error:stream.streamError];
                } else {
                    DDLogVerbose(@"[MQTTDecoder] read %ld %ld", (long)toRead, (long)n);
                    [self.dataBuffer appendBytes:buffer length:n];
                }
            }
            if (self.dataBuffer.length == self.length + self.offset) {
                DDLogVerbose(@"[MQTTDecoder] received (%lu)=%@...", (unsigned long)self.dataBuffer.length,
                                    [self.dataBuffer subdataWithRange:NSMakeRange(0, MIN(256, self.dataBuffer.length))]);
                [self.delegate decoder:self didReceiveMessage:self.dataBuffer];
                self.dataBuffer = nil;
                self.state = MQTTDecoderStateDecodingHeader;
            } else {
                DDLogWarn(@"[MQTTDecoder] oops received (%lu)=%@...", (unsigned long)self.dataBuffer.length,
                             [self.dataBuffer subdataWithRange:NSMakeRange(0, MIN(256, self.dataBuffer.length))]);
            }
        }
    }
    
    if (eventCode & NSStreamEventHasSpaceAvailable) {
        DDLogVerbose(@"[MQTTDecoder] NSStreamEventHasSpaceAvailable");
    }
    
    if (eventCode & NSStreamEventEndEncountered) {
        DDLogVerbose(@"[MQTTDecoder] NSStreamEventEndEncountered");
        
        if (self.streams) {
            [stream setDelegate:nil];
            [stream close];
            [self.streams removeObject:stream];
            if (self.streams.count) {
                NSInputStream *stream = (self.streams)[0];
                [stream open];
            }
        }
    }
    
    if (eventCode & NSStreamEventErrorOccurred) {
        DDLogVerbose(@"[MQTTDecoder] NSStreamEventErrorOccurred");
        
        self.state = MQTTDecoderStateConnectionError;
        NSError *error = stream.streamError;
        if (self.streams) {
            [self.streams removeObject:stream];
            if (self.streams.count) {
                NSInputStream *stream = (self.streams)[0];
                [stream open];
            }
        }
        [self.delegate decoder:self handleEvent:MQTTDecoderEventConnectionError error:error];
    }
}

@end