mirror of
https://github.com/smallmain/cocos-enhance-kit.git
synced 2025-01-14 06:51:08 +00:00
1639 lines
55 KiB
Objective-C
1639 lines
55 KiB
Objective-C
//
|
|
// Copyright 2012 Square Inc.
|
|
// Portions Copyright (c) 2016-present, Facebook, Inc.
|
|
//
|
|
// All rights reserved.
|
|
//
|
|
// This source code is licensed under the BSD-style license found in the
|
|
// LICENSE file in the root directory of this source tree. An additional grant
|
|
// of patent rights can be found in the PATENTS file in the same directory.
|
|
//
|
|
|
|
#import "SRWebSocket.h"
|
|
|
|
#if TARGET_OS_IPHONE
|
|
#define HAS_ICU
|
|
#endif
|
|
|
|
#ifdef HAS_ICU
|
|
#import <unicode/utf8.h>
|
|
#endif
|
|
|
|
#import <libkern/OSAtomic.h>
|
|
|
|
#import "SRDelegateController.h"
|
|
#import "SRIOConsumer.h"
|
|
#import "SRIOConsumerPool.h"
|
|
#import "SRHash.h"
|
|
#import "SRURLUtilities.h"
|
|
#import "SRError.h"
|
|
#import "NSURLRequest+SRWebSocket.h"
|
|
#import "NSRunLoop+SRWebSocket.h"
|
|
#import "SRProxyConnect.h"
|
|
#import "SRSecurityPolicy.h"
|
|
#import "SRHTTPConnectMessage.h"
|
|
#import "SRRandom.h"
|
|
#import "SRLog.h"
|
|
#import "SRMutex.h"
|
|
#import "SRSIMDHelpers.h"
|
|
#import "NSURLRequest+SRWebSocketPrivate.h"
|
|
#import "NSRunLoop+SRWebSocketPrivate.h"
|
|
#import "SRConstants.h"
|
|
|
|
#if !__has_feature(objc_arc)
|
|
#error SocketRocket must be compiled with ARC enabled
|
|
#endif
|
|
|
|
__attribute__((used)) static void importCategories()
|
|
{
|
|
import_NSURLRequest_SRWebSocket();
|
|
import_NSRunLoop_SRWebSocket();
|
|
}
|
|
|
|
typedef struct {
|
|
BOOL fin;
|
|
// BOOL rsv1;
|
|
// BOOL rsv2;
|
|
// BOOL rsv3;
|
|
uint8_t opcode;
|
|
BOOL masked;
|
|
uint64_t payload_length;
|
|
} frame_header;
|
|
|
|
static NSString *const SRWebSocketAppendToSecKeyString = @"258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
|
|
|
|
static inline int32_t validate_dispatch_data_partial_string(NSData *data);
|
|
|
|
static uint8_t const SRWebSocketProtocolVersion = 13;
|
|
|
|
NSString *const SRWebSocketErrorDomain = @"SRWebSocketErrorDomain";
|
|
NSString *const SRHTTPResponseErrorKey = @"HTTPResponseStatusCode";
|
|
|
|
@interface SRWebSocket () <NSStreamDelegate>
|
|
|
|
@property (atomic, assign, readwrite) SRReadyState readyState;
|
|
|
|
// Specifies whether SSL trust chain should NOT be evaluated.
|
|
// By default this flag is set to NO, meaning only secure SSL connections are allowed.
|
|
// For DEBUG builds this flag is ignored, and SSL connections are allowed regardless
|
|
// of the certificate trust configuration
|
|
@property (nonatomic, assign, readwrite) BOOL allowsUntrustedSSLCertificates;
|
|
|
|
@property (nonatomic, strong, readonly) SRDelegateController *delegateController;
|
|
|
|
@end
|
|
|
|
@implementation SRWebSocket {
|
|
SRMutex _kvoLock;
|
|
OSSpinLock _propertyLock;
|
|
|
|
dispatch_queue_t _workQueue;
|
|
NSMutableArray<SRIOConsumer *> *_consumers;
|
|
|
|
NSInputStream *_inputStream;
|
|
NSOutputStream *_outputStream;
|
|
|
|
dispatch_data_t _readBuffer;
|
|
NSUInteger _readBufferOffset;
|
|
|
|
dispatch_data_t _outputBuffer;
|
|
NSUInteger _outputBufferOffset;
|
|
|
|
uint8_t _currentFrameOpcode;
|
|
size_t _currentFrameCount;
|
|
size_t _readOpCount;
|
|
uint32_t _currentStringScanPosition;
|
|
NSMutableData *_currentFrameData;
|
|
|
|
NSString *_closeReason;
|
|
|
|
NSString *_secKey;
|
|
|
|
SRSecurityPolicy *_securityPolicy;
|
|
BOOL _requestRequiresSSL;
|
|
BOOL _streamSecurityValidated;
|
|
|
|
uint8_t _currentReadMaskKey[4];
|
|
size_t _currentReadMaskOffset;
|
|
|
|
BOOL _closeWhenFinishedWriting;
|
|
BOOL _failed;
|
|
|
|
NSURLRequest *_urlRequest;
|
|
|
|
BOOL _sentClose;
|
|
BOOL _didFail;
|
|
BOOL _cleanupScheduled;
|
|
int _closeCode;
|
|
|
|
BOOL _isPumping;
|
|
|
|
NSMutableSet<NSArray *> *_scheduledRunloops; // Set<[RunLoop, Mode]>. TODO: (nlutsenko) Fix clowntown
|
|
|
|
// We use this to retain ourselves.
|
|
__strong SRWebSocket *_selfRetain;
|
|
|
|
NSArray<NSString *> *_requestedProtocols;
|
|
SRIOConsumerPool *_consumerPool;
|
|
|
|
// proxy support
|
|
SRProxyConnect *_proxyConnect;
|
|
}
|
|
|
|
@synthesize readyState = _readyState;
|
|
|
|
///--------------------------------------
|
|
#pragma mark - Init
|
|
///--------------------------------------
|
|
|
|
- (instancetype)initWithURLRequest:(NSURLRequest *)request protocols:(NSArray<NSString *> *)protocols securityPolicy:(SRSecurityPolicy *)securityPolicy
|
|
{
|
|
self = [super init];
|
|
if (!self) return self;
|
|
|
|
assert(request.URL);
|
|
_url = request.URL;
|
|
_urlRequest = request;
|
|
_requestedProtocols = [protocols copy];
|
|
_securityPolicy = securityPolicy;
|
|
_requestRequiresSSL = SRURLRequiresSSL(_url);
|
|
|
|
_readyState = SR_CONNECTING;
|
|
|
|
_propertyLock = OS_SPINLOCK_INIT;
|
|
_kvoLock = SRMutexInitRecursive();
|
|
_workQueue = dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL);
|
|
|
|
// Going to set a specific on the queue so we can validate we're on the work queue
|
|
dispatch_queue_set_specific(_workQueue, (__bridge void *)self, (__bridge void *)(_workQueue), NULL);
|
|
|
|
_delegateController = [[SRDelegateController alloc] init];
|
|
|
|
_readBuffer = dispatch_data_empty;
|
|
_outputBuffer = dispatch_data_empty;
|
|
|
|
_currentFrameData = [[NSMutableData alloc] init];
|
|
|
|
_consumers = [[NSMutableArray alloc] init];
|
|
|
|
_consumerPool = [[SRIOConsumerPool alloc] init];
|
|
|
|
_scheduledRunloops = [[NSMutableSet alloc] init];
|
|
|
|
return self;
|
|
}
|
|
|
|
- (instancetype)initWithURLRequest:(NSURLRequest *)request protocols:(NSArray<NSString *> *)protocols allowsUntrustedSSLCertificates:(BOOL)allowsUntrustedSSLCertificates
|
|
{
|
|
SRSecurityPolicy *securityPolicy;
|
|
NSArray *pinnedCertificates = request.SR_SSLPinnedCertificates;
|
|
if (pinnedCertificates) {
|
|
securityPolicy = [SRSecurityPolicy pinnningPolicyWithCertificates:pinnedCertificates];
|
|
} else {
|
|
BOOL certificateChainValidationEnabled = !allowsUntrustedSSLCertificates;
|
|
securityPolicy = [[SRSecurityPolicy alloc] initWithCertificateChainValidationEnabled:certificateChainValidationEnabled];
|
|
}
|
|
|
|
return [self initWithURLRequest:request protocols:protocols securityPolicy:securityPolicy];
|
|
}
|
|
|
|
- (instancetype)initWithURLRequest:(NSURLRequest *)request securityPolicy:(SRSecurityPolicy *)securityPolicy
|
|
{
|
|
return [self initWithURLRequest:request protocols:nil securityPolicy:securityPolicy];
|
|
}
|
|
|
|
- (instancetype)initWithURLRequest:(NSURLRequest *)request protocols:(NSArray<NSString *> *)protocols
|
|
{
|
|
return [self initWithURLRequest:request protocols:protocols allowsUntrustedSSLCertificates:NO];
|
|
}
|
|
|
|
- (instancetype)initWithURLRequest:(NSURLRequest *)request
|
|
{
|
|
return [self initWithURLRequest:request protocols:nil];
|
|
}
|
|
|
|
- (instancetype)initWithURL:(NSURL *)url;
|
|
{
|
|
return [self initWithURL:url protocols:nil];
|
|
}
|
|
|
|
- (instancetype)initWithURL:(NSURL *)url protocols:(NSArray<NSString *> *)protocols;
|
|
{
|
|
return [self initWithURL:url protocols:protocols allowsUntrustedSSLCertificates:NO];
|
|
}
|
|
|
|
- (instancetype)initWithURL:(NSURL *)url securityPolicy:(SRSecurityPolicy *)securityPolicy
|
|
{
|
|
NSURLRequest *request = [NSURLRequest requestWithURL:url];
|
|
return [self initWithURLRequest:request protocols:nil securityPolicy:securityPolicy];
|
|
}
|
|
|
|
- (instancetype)initWithURL:(NSURL *)url protocols:(NSArray<NSString *> *)protocols allowsUntrustedSSLCertificates:(BOOL)allowsUntrustedSSLCertificates
|
|
{
|
|
NSURLRequest *request = [NSURLRequest requestWithURL:url];
|
|
return [self initWithURLRequest:request protocols:protocols allowsUntrustedSSLCertificates:allowsUntrustedSSLCertificates];
|
|
}
|
|
|
|
- (void)assertOnWorkQueue;
|
|
{
|
|
assert(dispatch_get_specific((__bridge void *)self) == (__bridge void *)_workQueue);
|
|
}
|
|
|
|
///--------------------------------------
|
|
#pragma mark - Dealloc
|
|
///--------------------------------------
|
|
|
|
- (void)dealloc
|
|
{
|
|
_inputStream.delegate = nil;
|
|
_outputStream.delegate = nil;
|
|
|
|
[_inputStream close];
|
|
[_outputStream close];
|
|
|
|
if (_receivedHTTPHeaders) {
|
|
CFRelease(_receivedHTTPHeaders);
|
|
_receivedHTTPHeaders = NULL;
|
|
}
|
|
|
|
SRMutexDestroy(_kvoLock);
|
|
}
|
|
|
|
///--------------------------------------
|
|
#pragma mark - Accessors
|
|
///--------------------------------------
|
|
|
|
#pragma mark readyState
|
|
|
|
- (void)setReadyState:(SRReadyState)readyState
|
|
{
|
|
@try {
|
|
SRMutexLock(_kvoLock);
|
|
if (_readyState != readyState) {
|
|
[self willChangeValueForKey:@"readyState"];
|
|
OSSpinLockLock(&_propertyLock);
|
|
_readyState = readyState;
|
|
OSSpinLockUnlock(&_propertyLock);
|
|
[self didChangeValueForKey:@"readyState"];
|
|
}
|
|
}
|
|
@finally {
|
|
SRMutexUnlock(_kvoLock);
|
|
}
|
|
}
|
|
|
|
- (SRReadyState)readyState
|
|
{
|
|
SRReadyState state = 0;
|
|
OSSpinLockLock(&_propertyLock);
|
|
state = _readyState;
|
|
OSSpinLockUnlock(&_propertyLock);
|
|
return state;
|
|
}
|
|
|
|
+ (BOOL)automaticallyNotifiesObserversOfReadyState {
|
|
return NO;
|
|
}
|
|
|
|
-(void)_onTimeout
|
|
{
|
|
if (self.readyState == SR_CONNECTING) {
|
|
NSError *error = SRErrorWithDomainCodeDescription(NSURLErrorDomain, NSURLErrorTimedOut, @"Timed out connecting to server.");
|
|
[self _failWithError:error];
|
|
}
|
|
}
|
|
|
|
///--------------------------------------
|
|
#pragma mark - Open / Close
|
|
///--------------------------------------
|
|
|
|
- (void)open
|
|
{
|
|
assert(_url);
|
|
NSAssert(self.readyState == SR_CONNECTING, @"Cannot call -(void)open on SRWebSocket more than once.");
|
|
|
|
_selfRetain = self;
|
|
|
|
if (_urlRequest.timeoutInterval > 0) {
|
|
[self performSelector:@selector(_onTimeout) withObject:nil afterDelay:_urlRequest.timeoutInterval];
|
|
}
|
|
|
|
_proxyConnect = [[SRProxyConnect alloc] initWithURL:_url];
|
|
|
|
__weak __typeof(self) wself = self;
|
|
[_proxyConnect openNetworkStreamWithCompletion:^(NSError *error, NSInputStream *readStream, NSOutputStream *writeStream) {
|
|
[wself _connectionDoneWithError:error readStream:readStream writeStream:writeStream];
|
|
}];
|
|
}
|
|
|
|
- (void)_connectionDoneWithError:(NSError *)error readStream:(NSInputStream *)readStream writeStream:(NSOutputStream *)writeStream
|
|
{
|
|
if (error != nil) {
|
|
[self _failWithError:error];
|
|
} else {
|
|
_outputStream = writeStream;
|
|
_inputStream = readStream;
|
|
|
|
_inputStream.delegate = self;
|
|
_outputStream.delegate = self;
|
|
[self _updateSecureStreamOptions];
|
|
|
|
if (!_scheduledRunloops.count) {
|
|
[self scheduleInRunLoop:[NSRunLoop SR_networkRunLoop] forMode:NSDefaultRunLoopMode];
|
|
}
|
|
|
|
// If we don't require SSL validation - consider that we connected.
|
|
// Otherwise `didConnect` is called when SSL validation finishes.
|
|
if (!_requestRequiresSSL) {
|
|
dispatch_async(_workQueue, ^{
|
|
[self didConnect];
|
|
});
|
|
}
|
|
}
|
|
// Schedule to run on a work queue, to make sure we don't run this inline and deallocate `self` inside `SRProxyConnect`.
|
|
// TODO: (nlutsenko) Find a better structure for this, maybe Bolts Tasks?
|
|
dispatch_async(_workQueue, ^{
|
|
_proxyConnect = nil;
|
|
});
|
|
}
|
|
|
|
- (BOOL)_checkHandshake:(CFHTTPMessageRef)httpMessage;
|
|
{
|
|
NSString *acceptHeader = CFBridgingRelease(CFHTTPMessageCopyHeaderFieldValue(httpMessage, CFSTR("Sec-WebSocket-Accept")));
|
|
|
|
if (acceptHeader == nil) {
|
|
return NO;
|
|
}
|
|
|
|
NSString *concattedString = [_secKey stringByAppendingString:SRWebSocketAppendToSecKeyString];
|
|
NSData *hashedString = SRSHA1HashFromString(concattedString);
|
|
NSString *expectedAccept = SRBase64EncodedStringFromData(hashedString);
|
|
return [acceptHeader isEqualToString:expectedAccept];
|
|
}
|
|
|
|
- (void)_HTTPHeadersDidFinish;
|
|
{
|
|
NSInteger responseCode = CFHTTPMessageGetResponseStatusCode(_receivedHTTPHeaders);
|
|
if (responseCode >= 400) {
|
|
SRDebugLog(@"Request failed with response code %d", responseCode);
|
|
NSError *error = SRHTTPErrorWithCodeDescription(responseCode, 2132,
|
|
[NSString stringWithFormat:@"Received bad response code from server: %d.",
|
|
(int)responseCode]);
|
|
[self _failWithError:error];
|
|
return;
|
|
}
|
|
|
|
if(![self _checkHandshake:_receivedHTTPHeaders]) {
|
|
NSError *error = SRErrorWithCodeDescription(2133, @"Invalid Sec-WebSocket-Accept response.");
|
|
[self _failWithError:error];
|
|
return;
|
|
}
|
|
|
|
NSString *negotiatedProtocol = CFBridgingRelease(CFHTTPMessageCopyHeaderFieldValue(_receivedHTTPHeaders, CFSTR("Sec-WebSocket-Protocol")));
|
|
if (negotiatedProtocol) {
|
|
// Make sure we requested the protocol
|
|
if ([_requestedProtocols indexOfObject:negotiatedProtocol] == NSNotFound) {
|
|
NSError *error = SRErrorWithCodeDescription(2133, @"Server specified Sec-WebSocket-Protocol that wasn't requested.");
|
|
[self _failWithError:error];
|
|
return;
|
|
}
|
|
|
|
_protocol = negotiatedProtocol;
|
|
}
|
|
|
|
self.readyState = SR_OPEN;
|
|
|
|
if (!_didFail) {
|
|
[self _readFrameNew];
|
|
}
|
|
|
|
[self.delegateController performDelegateBlock:^(id<SRWebSocketDelegate> _Nullable delegate, SRDelegateAvailableMethods availableMethods) {
|
|
if (availableMethods.didOpen) {
|
|
[delegate webSocketDidOpen:self];
|
|
}
|
|
}];
|
|
}
|
|
|
|
|
|
- (void)_readHTTPHeader;
|
|
{
|
|
if (_receivedHTTPHeaders == NULL) {
|
|
_receivedHTTPHeaders = CFHTTPMessageCreateEmpty(NULL, NO);
|
|
}
|
|
|
|
// Uses weak self object in the block, otherwise Consumers will retain SRWebSocket instance,
|
|
// and SRWebSocket instance also hold consumers, cycle reference will occur.
|
|
__weak __typeof(self) wself = self;
|
|
|
|
[self _readUntilHeaderCompleteWithCallback:^(SRWebSocket *socket, NSData *data) {
|
|
|
|
__strong __typeof(wself) sself = wself;
|
|
if (sself == nil)
|
|
return;
|
|
|
|
CFHTTPMessageAppendBytes(sself.receivedHTTPHeaders, (const UInt8 *)data.bytes, data.length);
|
|
|
|
if (CFHTTPMessageIsHeaderComplete(sself.receivedHTTPHeaders)) {
|
|
SRDebugLog(@"Finished reading headers %@", CFBridgingRelease(CFHTTPMessageCopyAllHeaderFields(sself.receivedHTTPHeaders)));
|
|
[sself _HTTPHeadersDidFinish];
|
|
} else {
|
|
[sself _readHTTPHeader];
|
|
}
|
|
}];
|
|
}
|
|
|
|
- (void)didConnect;
|
|
{
|
|
SRDebugLog(@"Connected");
|
|
|
|
_secKey = SRBase64EncodedStringFromData(SRRandomData(16));
|
|
assert([_secKey length] == 24);
|
|
|
|
CFHTTPMessageRef message = SRHTTPConnectMessageCreate(_urlRequest,
|
|
_secKey,
|
|
SRWebSocketProtocolVersion,
|
|
self.requestCookies,
|
|
_requestedProtocols);
|
|
|
|
NSData *messageData = CFBridgingRelease(CFHTTPMessageCopySerializedMessage(message));
|
|
|
|
CFRelease(message);
|
|
|
|
[self _writeData:messageData];
|
|
[self _readHTTPHeader];
|
|
}
|
|
|
|
- (void)_updateSecureStreamOptions
|
|
{
|
|
if (_requestRequiresSSL) {
|
|
SRDebugLog(@"Setting up security for streams.");
|
|
[_securityPolicy updateSecurityOptionsInStream:_inputStream];
|
|
[_securityPolicy updateSecurityOptionsInStream:_outputStream];
|
|
}
|
|
|
|
NSString *networkServiceType = SRStreamNetworkServiceTypeFromURLRequest(_urlRequest);
|
|
if (networkServiceType != nil) {
|
|
[_inputStream setProperty:networkServiceType forKey:NSStreamNetworkServiceType];
|
|
[_outputStream setProperty:networkServiceType forKey:NSStreamNetworkServiceType];
|
|
}
|
|
}
|
|
|
|
- (void)scheduleInRunLoop:(NSRunLoop *)aRunLoop forMode:(NSString *)mode;
|
|
{
|
|
[_outputStream scheduleInRunLoop:aRunLoop forMode:mode];
|
|
[_inputStream scheduleInRunLoop:aRunLoop forMode:mode];
|
|
|
|
[_scheduledRunloops addObject:@[aRunLoop, mode]];
|
|
}
|
|
|
|
- (void)unscheduleFromRunLoop:(NSRunLoop *)aRunLoop forMode:(NSString *)mode;
|
|
{
|
|
[_outputStream removeFromRunLoop:aRunLoop forMode:mode];
|
|
[_inputStream removeFromRunLoop:aRunLoop forMode:mode];
|
|
|
|
[_scheduledRunloops removeObject:@[aRunLoop, mode]];
|
|
}
|
|
|
|
- (void)close;
|
|
{
|
|
[self closeWithCode:SRStatusCodeNormal reason:nil];
|
|
}
|
|
|
|
- (void)closeWithCode:(NSInteger)code reason:(NSString *)reason;
|
|
{
|
|
assert(code);
|
|
dispatch_async(_workQueue, ^{
|
|
if (self.readyState == SR_CLOSING || self.readyState == SR_CLOSED) {
|
|
return;
|
|
}
|
|
|
|
BOOL wasConnecting = self.readyState == SR_CONNECTING;
|
|
|
|
self.readyState = SR_CLOSING;
|
|
|
|
SRDebugLog(@"Closing with code %d reason %@", code, reason);
|
|
|
|
if (wasConnecting) {
|
|
[self closeConnection];
|
|
return;
|
|
}
|
|
|
|
size_t maxMsgSize = [reason maximumLengthOfBytesUsingEncoding:NSUTF8StringEncoding];
|
|
NSMutableData *mutablePayload = [[NSMutableData alloc] initWithLength:sizeof(uint16_t) + maxMsgSize];
|
|
NSData *payload = mutablePayload;
|
|
|
|
((uint16_t *)mutablePayload.mutableBytes)[0] = CFSwapInt16BigToHost((uint16_t)code);
|
|
|
|
if (reason) {
|
|
NSRange remainingRange = {0};
|
|
|
|
NSUInteger usedLength = 0;
|
|
|
|
BOOL success = [reason getBytes:(char *)mutablePayload.mutableBytes + sizeof(uint16_t) maxLength:payload.length - sizeof(uint16_t) usedLength:&usedLength encoding:NSUTF8StringEncoding options:NSStringEncodingConversionExternalRepresentation range:NSMakeRange(0, reason.length) remainingRange:&remainingRange];
|
|
#pragma unused (success)
|
|
|
|
assert(success);
|
|
assert(remainingRange.length == 0);
|
|
|
|
if (usedLength != maxMsgSize) {
|
|
payload = [payload subdataWithRange:NSMakeRange(0, usedLength + sizeof(uint16_t))];
|
|
}
|
|
}
|
|
|
|
|
|
[self _sendFrameWithOpcode:SROpCodeConnectionClose data:payload];
|
|
});
|
|
}
|
|
|
|
- (void)_closeWithProtocolError:(NSString *)message;
|
|
{
|
|
// Need to shunt this on the _callbackQueue first to see if they received any messages
|
|
[self.delegateController performDelegateQueueBlock:^{
|
|
[self closeWithCode:SRStatusCodeProtocolError reason:message];
|
|
dispatch_async(_workQueue, ^{
|
|
[self closeConnection];
|
|
});
|
|
}];
|
|
}
|
|
|
|
- (void)_failWithError:(NSError *)error;
|
|
{
|
|
dispatch_async(_workQueue, ^{
|
|
if (self.readyState != SR_CLOSED) {
|
|
_failed = YES;
|
|
[self.delegateController performDelegateBlock:^(id<SRWebSocketDelegate> _Nullable delegate, SRDelegateAvailableMethods availableMethods) {
|
|
if (availableMethods.didFailWithError) {
|
|
[delegate webSocket:self didFailWithError:error];
|
|
}
|
|
}];
|
|
|
|
self.readyState = SR_CLOSED;
|
|
|
|
SRDebugLog(@"Failing with error %@", error.localizedDescription);
|
|
|
|
[self closeConnection];
|
|
[self _scheduleCleanup];
|
|
}
|
|
});
|
|
}
|
|
|
|
- (void)_writeData:(NSData *)data;
|
|
{
|
|
[self assertOnWorkQueue];
|
|
|
|
if (_closeWhenFinishedWriting) {
|
|
return;
|
|
}
|
|
|
|
__block NSData *strongData = data;
|
|
dispatch_data_t newData = dispatch_data_create(data.bytes, data.length, nil, ^{
|
|
strongData = nil;
|
|
});
|
|
_outputBuffer = dispatch_data_create_concat(_outputBuffer, newData);
|
|
[self _pumpWriting];
|
|
}
|
|
|
|
- (void)send:(nullable id)message
|
|
{
|
|
if (!message) {
|
|
[self sendData:nil error:nil]; // Send Data, but it doesn't matter since we are going to send the same text frame with 0 length.
|
|
} else if ([message isKindOfClass:[NSString class]]) {
|
|
[self sendString:message error:nil];
|
|
} else if ([message isKindOfClass:[NSData class]]) {
|
|
[self sendData:message error:nil];
|
|
} else {
|
|
NSAssert(NO, @"Unrecognized message. Not able to send anything other than a String or NSData.");
|
|
}
|
|
}
|
|
|
|
- (BOOL)sendString:(NSString *)string error:(NSError **)error
|
|
{
|
|
if (self.readyState != SR_OPEN) {
|
|
NSString *message = @"Invalid State: Cannot call `sendString:error:` until connection is open.";
|
|
if (error) {
|
|
*error = SRErrorWithCodeDescription(2134, message);
|
|
}
|
|
SRDebugLog(message);
|
|
return NO;
|
|
}
|
|
|
|
string = [string copy];
|
|
dispatch_async(_workQueue, ^{
|
|
[self _sendFrameWithOpcode:SROpCodeTextFrame data:[string dataUsingEncoding:NSUTF8StringEncoding]];
|
|
});
|
|
return YES;
|
|
}
|
|
|
|
- (BOOL)sendData:(nullable NSData *)data error:(NSError **)error
|
|
{
|
|
data = [data copy];
|
|
return [self sendDataNoCopy:data error:error];
|
|
}
|
|
|
|
- (BOOL)sendDataNoCopy:(nullable NSData *)data error:(NSError **)error
|
|
{
|
|
if (self.readyState != SR_OPEN) {
|
|
NSString *message = @"Invalid State: Cannot call `sendDataNoCopy:error:` until connection is open.";
|
|
if (error) {
|
|
*error = SRErrorWithCodeDescription(2134, message);
|
|
}
|
|
SRDebugLog(message);
|
|
return NO;
|
|
}
|
|
|
|
dispatch_async(_workQueue, ^{
|
|
if (data) {
|
|
[self _sendFrameWithOpcode:SROpCodeBinaryFrame data:data];
|
|
} else {
|
|
[self _sendFrameWithOpcode:SROpCodeTextFrame data:nil];
|
|
}
|
|
});
|
|
return YES;
|
|
}
|
|
|
|
- (BOOL)sendPing:(nullable NSData *)data error:(NSError **)error
|
|
{
|
|
if (self.readyState != SR_OPEN) {
|
|
NSString *message = @"Invalid State: Cannot call `sendPing:error:` until connection is open.";
|
|
if (error) {
|
|
*error = SRErrorWithCodeDescription(2134, message);
|
|
}
|
|
SRDebugLog(message);
|
|
return NO;
|
|
}
|
|
|
|
data = [data copy] ?: [NSData data]; // It's okay for a ping to be empty
|
|
dispatch_async(_workQueue, ^{
|
|
[self _sendFrameWithOpcode:SROpCodePing data:data];
|
|
});
|
|
return YES;
|
|
}
|
|
|
|
- (void)_handlePingWithData:(nullable NSData *)data
|
|
{
|
|
// Need to pingpong this off _callbackQueue first to make sure messages happen in order
|
|
[self.delegateController performDelegateBlock:^(id<SRWebSocketDelegate> _Nullable delegate, SRDelegateAvailableMethods availableMethods) {
|
|
if (availableMethods.didReceivePing) {
|
|
[delegate webSocket:self didReceivePingWithData:data];
|
|
}
|
|
dispatch_async(_workQueue, ^{
|
|
[self _sendFrameWithOpcode:SROpCodePong data:data];
|
|
});
|
|
}];
|
|
}
|
|
|
|
- (void)handlePong:(NSData *)pongData;
|
|
{
|
|
SRDebugLog(@"Received pong");
|
|
[self.delegateController performDelegateBlock:^(id<SRWebSocketDelegate> _Nullable delegate, SRDelegateAvailableMethods availableMethods) {
|
|
if (availableMethods.didReceivePong) {
|
|
[delegate webSocket:self didReceivePong:pongData];
|
|
}
|
|
}];
|
|
}
|
|
|
|
|
|
static inline BOOL closeCodeIsValid(int closeCode) {
|
|
if (closeCode < 1000) {
|
|
return NO;
|
|
}
|
|
|
|
if (closeCode >= 1000 && closeCode <= 1011) {
|
|
if (closeCode == 1004 ||
|
|
closeCode == 1005 ||
|
|
closeCode == 1006) {
|
|
return NO;
|
|
}
|
|
return YES;
|
|
}
|
|
|
|
if (closeCode >= 3000 && closeCode <= 3999) {
|
|
return YES;
|
|
}
|
|
|
|
if (closeCode >= 4000 && closeCode <= 4999) {
|
|
return YES;
|
|
}
|
|
|
|
return NO;
|
|
}
|
|
|
|
// Note from RFC:
|
|
//
|
|
// If there is a body, the first two
|
|
// bytes of the body MUST be a 2-byte unsigned integer (in network byte
|
|
// order) representing a status code with value /code/ defined in
|
|
// Section 7.4. Following the 2-byte integer the body MAY contain UTF-8
|
|
// encoded data with value /reason/, the interpretation of which is not
|
|
// defined by this specification.
|
|
|
|
- (void)handleCloseWithData:(NSData *)data;
|
|
{
|
|
size_t dataSize = data.length;
|
|
__block uint16_t closeCode = 0;
|
|
|
|
SRDebugLog(@"Received close frame");
|
|
|
|
if (dataSize == 1) {
|
|
// TODO handle error
|
|
[self _closeWithProtocolError:@"Payload for close must be larger than 2 bytes"];
|
|
return;
|
|
} else if (dataSize >= 2) {
|
|
[data getBytes:&closeCode length:sizeof(closeCode)];
|
|
_closeCode = CFSwapInt16BigToHost(closeCode);
|
|
if (!closeCodeIsValid(_closeCode)) {
|
|
[self _closeWithProtocolError:[NSString stringWithFormat:@"Cannot have close code of %d", _closeCode]];
|
|
return;
|
|
}
|
|
if (dataSize > 2) {
|
|
_closeReason = [[NSString alloc] initWithData:[data subdataWithRange:NSMakeRange(2, dataSize - 2)] encoding:NSUTF8StringEncoding];
|
|
if (!_closeReason) {
|
|
[self _closeWithProtocolError:@"Close reason MUST be valid UTF-8"];
|
|
return;
|
|
}
|
|
}
|
|
} else {
|
|
_closeCode = SRStatusNoStatusReceived;
|
|
}
|
|
|
|
[self assertOnWorkQueue];
|
|
|
|
if (self.readyState == SR_OPEN) {
|
|
[self closeWithCode:1000 reason:nil];
|
|
}
|
|
dispatch_async(_workQueue, ^{
|
|
[self closeConnection];
|
|
});
|
|
}
|
|
|
|
- (void)closeConnection;
|
|
{
|
|
[self assertOnWorkQueue];
|
|
SRDebugLog(@"Trying to disconnect");
|
|
_closeWhenFinishedWriting = YES;
|
|
[self _pumpWriting];
|
|
}
|
|
|
|
- (void)_handleFrameWithData:(NSData *)frameData opCode:(SROpCode)opcode
|
|
{
|
|
// Check that the current data is valid UTF8
|
|
|
|
BOOL isControlFrame = (opcode == SROpCodePing || opcode == SROpCodePong || opcode == SROpCodeConnectionClose);
|
|
if (isControlFrame) {
|
|
//frameData will be copied before passing to handlers
|
|
//otherwise there can be misbehaviours when value at the pointer is changed
|
|
frameData = [frameData copy];
|
|
|
|
dispatch_async(_workQueue, ^{
|
|
[self _readFrameContinue];
|
|
});
|
|
} else {
|
|
[self _readFrameNew];
|
|
}
|
|
|
|
switch (opcode) {
|
|
case SROpCodeTextFrame: {
|
|
NSString *string = [[NSString alloc] initWithData:frameData encoding:NSUTF8StringEncoding];
|
|
if (!string && frameData) {
|
|
[self closeWithCode:SRStatusCodeInvalidUTF8 reason:@"Text frames must be valid UTF-8."];
|
|
dispatch_async(_workQueue, ^{
|
|
[self closeConnection];
|
|
});
|
|
return;
|
|
}
|
|
SRDebugLog(@"Received text message.");
|
|
[self.delegateController performDelegateBlock:^(id<SRWebSocketDelegate> _Nullable delegate, SRDelegateAvailableMethods availableMethods) {
|
|
// Don't convert into string - iff `delegate` tells us not to. Otherwise - create UTF8 string and handle that.
|
|
if (availableMethods.shouldConvertTextFrameToString && ![delegate webSocketShouldConvertTextFrameToString:self]) {
|
|
if (availableMethods.didReceiveMessage) {
|
|
[delegate webSocket:self didReceiveMessage:frameData];
|
|
}
|
|
if (availableMethods.didReceiveMessageWithData) {
|
|
[delegate webSocket:self didReceiveMessageWithData:frameData];
|
|
}
|
|
} else {
|
|
if (availableMethods.didReceiveMessage) {
|
|
[delegate webSocket:self didReceiveMessage:string];
|
|
}
|
|
if (availableMethods.didReceiveMessageWithString) {
|
|
[delegate webSocket:self didReceiveMessageWithString:string];
|
|
}
|
|
}
|
|
}];
|
|
break;
|
|
}
|
|
case SROpCodeBinaryFrame: {
|
|
SRDebugLog(@"Received data message.");
|
|
[self.delegateController performDelegateBlock:^(id<SRWebSocketDelegate> _Nullable delegate, SRDelegateAvailableMethods availableMethods) {
|
|
if (availableMethods.didReceiveMessage) {
|
|
[delegate webSocket:self didReceiveMessage:frameData];
|
|
}
|
|
if (availableMethods.didReceiveMessageWithData) {
|
|
[delegate webSocket:self didReceiveMessageWithData:frameData];
|
|
}
|
|
}];
|
|
}
|
|
break;
|
|
case SROpCodeConnectionClose:
|
|
[self handleCloseWithData:frameData];
|
|
break;
|
|
case SROpCodePing:
|
|
[self _handlePingWithData:frameData];
|
|
break;
|
|
case SROpCodePong:
|
|
[self handlePong:frameData];
|
|
break;
|
|
default:
|
|
[self _closeWithProtocolError:[NSString stringWithFormat:@"Unknown opcode %ld", (long)opcode]];
|
|
// TODO: Handle invalid opcode
|
|
break;
|
|
}
|
|
}
|
|
|
|
- (void)_handleFrameHeader:(frame_header)frame_header curData:(NSData *)curData;
|
|
{
|
|
assert(frame_header.opcode != 0);
|
|
|
|
if (self.readyState == SR_CLOSED) {
|
|
return;
|
|
}
|
|
|
|
|
|
BOOL isControlFrame = (frame_header.opcode == SROpCodePing || frame_header.opcode == SROpCodePong || frame_header.opcode == SROpCodeConnectionClose);
|
|
|
|
if (isControlFrame && !frame_header.fin) {
|
|
[self _closeWithProtocolError:@"Fragmented control frames not allowed"];
|
|
return;
|
|
}
|
|
|
|
if (isControlFrame && frame_header.payload_length >= 126) {
|
|
[self _closeWithProtocolError:@"Control frames cannot have payloads larger than 126 bytes"];
|
|
return;
|
|
}
|
|
|
|
if (!isControlFrame) {
|
|
_currentFrameOpcode = frame_header.opcode;
|
|
_currentFrameCount += 1;
|
|
}
|
|
|
|
if (frame_header.payload_length == 0) {
|
|
if (isControlFrame) {
|
|
[self _handleFrameWithData:curData opCode:frame_header.opcode];
|
|
} else {
|
|
if (frame_header.fin) {
|
|
[self _handleFrameWithData:_currentFrameData opCode:frame_header.opcode];
|
|
} else {
|
|
// TODO add assert that opcode is not a control;
|
|
[self _readFrameContinue];
|
|
}
|
|
}
|
|
} else {
|
|
assert(frame_header.payload_length <= SIZE_T_MAX);
|
|
[self _addConsumerWithDataLength:(size_t)frame_header.payload_length callback:^(SRWebSocket *sself, NSData *newData) {
|
|
if (isControlFrame) {
|
|
[sself _handleFrameWithData:newData opCode:frame_header.opcode];
|
|
} else {
|
|
if (frame_header.fin) {
|
|
[sself _handleFrameWithData:sself->_currentFrameData opCode:frame_header.opcode];
|
|
} else {
|
|
// TODO add assert that opcode is not a control;
|
|
[sself _readFrameContinue];
|
|
}
|
|
}
|
|
} readToCurrentFrame:!isControlFrame unmaskBytes:frame_header.masked];
|
|
}
|
|
}
|
|
|
|
/* From RFC:
|
|
|
|
0 1 2 3
|
|
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
|
|
+-+-+-+-+-------+-+-------------+-------------------------------+
|
|
|F|R|R|R| opcode|M| Payload len | Extended payload length |
|
|
|I|S|S|S| (4) |A| (7) | (16/64) |
|
|
|N|V|V|V| |S| | (if payload len==126/127) |
|
|
| |1|2|3| |K| | |
|
|
+-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
|
|
| Extended payload length continued, if payload len == 127 |
|
|
+ - - - - - - - - - - - - - - - +-------------------------------+
|
|
| |Masking-key, if MASK set to 1 |
|
|
+-------------------------------+-------------------------------+
|
|
| Masking-key (continued) | Payload Data |
|
|
+-------------------------------- - - - - - - - - - - - - - - - +
|
|
: Payload Data continued ... :
|
|
+ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
|
|
| Payload Data continued ... |
|
|
+---------------------------------------------------------------+
|
|
*/
|
|
|
|
static const uint8_t SRFinMask = 0x80;
|
|
static const uint8_t SROpCodeMask = 0x0F;
|
|
static const uint8_t SRRsvMask = 0x70;
|
|
static const uint8_t SRMaskMask = 0x80;
|
|
static const uint8_t SRPayloadLenMask = 0x7F;
|
|
|
|
|
|
- (void)_readFrameContinue;
|
|
{
|
|
assert((_currentFrameCount == 0 && _currentFrameOpcode == 0) || (_currentFrameCount > 0 && _currentFrameOpcode > 0));
|
|
|
|
[self _addConsumerWithDataLength:2 callback:^(SRWebSocket *sself, NSData *data) {
|
|
__block frame_header header = {0};
|
|
|
|
const uint8_t *headerBuffer = data.bytes;
|
|
assert(data.length >= 2);
|
|
|
|
if (headerBuffer[0] & SRRsvMask) {
|
|
[sself _closeWithProtocolError:@"Server used RSV bits"];
|
|
return;
|
|
}
|
|
|
|
uint8_t receivedOpcode = (SROpCodeMask & headerBuffer[0]);
|
|
|
|
BOOL isControlFrame = (receivedOpcode == SROpCodePing || receivedOpcode == SROpCodePong || receivedOpcode == SROpCodeConnectionClose);
|
|
|
|
if (!isControlFrame && receivedOpcode != 0 && sself->_currentFrameCount > 0) {
|
|
[sself _closeWithProtocolError:@"all data frames after the initial data frame must have opcode 0"];
|
|
return;
|
|
}
|
|
|
|
if (receivedOpcode == 0 && sself->_currentFrameCount == 0) {
|
|
[sself _closeWithProtocolError:@"cannot continue a message"];
|
|
return;
|
|
}
|
|
|
|
header.opcode = receivedOpcode == 0 ? sself->_currentFrameOpcode : receivedOpcode;
|
|
|
|
header.fin = !!(SRFinMask & headerBuffer[0]);
|
|
|
|
|
|
header.masked = !!(SRMaskMask & headerBuffer[1]);
|
|
header.payload_length = SRPayloadLenMask & headerBuffer[1];
|
|
|
|
headerBuffer = NULL;
|
|
|
|
if (header.masked) {
|
|
[sself _closeWithProtocolError:@"Client must receive unmasked data"];
|
|
return;
|
|
}
|
|
|
|
size_t extra_bytes_needed = header.masked ? sizeof(_currentReadMaskKey) : 0;
|
|
|
|
if (header.payload_length == 126) {
|
|
extra_bytes_needed += sizeof(uint16_t);
|
|
} else if (header.payload_length == 127) {
|
|
extra_bytes_needed += sizeof(uint64_t);
|
|
}
|
|
|
|
if (extra_bytes_needed == 0) {
|
|
[sself _handleFrameHeader:header curData:sself->_currentFrameData];
|
|
} else {
|
|
[sself _addConsumerWithDataLength:extra_bytes_needed callback:^(SRWebSocket *eself, NSData *edata) {
|
|
size_t mapped_size = edata.length;
|
|
#pragma unused (mapped_size)
|
|
const void *mapped_buffer = edata.bytes;
|
|
size_t offset = 0;
|
|
|
|
if (header.payload_length == 126) {
|
|
assert(mapped_size >= sizeof(uint16_t));
|
|
uint16_t payloadLength = 0;
|
|
memcpy(&payloadLength, mapped_buffer, sizeof(uint16_t));
|
|
payloadLength = CFSwapInt16BigToHost(payloadLength);
|
|
|
|
header.payload_length = payloadLength;
|
|
offset += sizeof(uint16_t);
|
|
} else if (header.payload_length == 127) {
|
|
assert(mapped_size >= sizeof(uint64_t));
|
|
uint64_t payloadLength = 0;
|
|
memcpy(&payloadLength, mapped_buffer, sizeof(uint64_t));
|
|
payloadLength = CFSwapInt64BigToHost(payloadLength);
|
|
|
|
header.payload_length = payloadLength;
|
|
offset += sizeof(uint64_t);
|
|
} else {
|
|
assert(header.payload_length < 126 && header.payload_length >= 0);
|
|
}
|
|
|
|
if (header.masked) {
|
|
assert(mapped_size >= sizeof(_currentReadMaskOffset) + offset);
|
|
memcpy(eself->_currentReadMaskKey, ((uint8_t *)mapped_buffer) + offset, sizeof(eself->_currentReadMaskKey));
|
|
}
|
|
|
|
[eself _handleFrameHeader:header curData:eself->_currentFrameData];
|
|
} readToCurrentFrame:NO unmaskBytes:NO];
|
|
}
|
|
} readToCurrentFrame:NO unmaskBytes:NO];
|
|
}
|
|
|
|
- (void)_readFrameNew;
|
|
{
|
|
dispatch_async(_workQueue, ^{
|
|
// Don't reset the length, since Apple doesn't guarantee that this will free the memory (and in tests on
|
|
// some platforms, it doesn't seem to, effectively causing a leak the size of the biggest frame so far).
|
|
_currentFrameData = [[NSMutableData alloc] init];
|
|
|
|
_currentFrameOpcode = 0;
|
|
_currentFrameCount = 0;
|
|
_readOpCount = 0;
|
|
_currentStringScanPosition = 0;
|
|
|
|
[self _readFrameContinue];
|
|
});
|
|
}
|
|
|
|
- (void)_pumpWriting;
|
|
{
|
|
[self assertOnWorkQueue];
|
|
|
|
NSUInteger dataLength = dispatch_data_get_size(_outputBuffer);
|
|
if (dataLength - _outputBufferOffset > 0 && _outputStream.hasSpaceAvailable) {
|
|
__block NSInteger bytesWritten = 0;
|
|
__block BOOL streamFailed = NO;
|
|
|
|
dispatch_data_t dataToSend = dispatch_data_create_subrange(_outputBuffer, _outputBufferOffset, dataLength - _outputBufferOffset);
|
|
dispatch_data_apply(dataToSend, ^bool(dispatch_data_t region, size_t offset, const void *buffer, size_t size) {
|
|
NSInteger sentLength = [_outputStream write:buffer maxLength:size];
|
|
if (sentLength == -1) {
|
|
streamFailed = YES;
|
|
return false;
|
|
}
|
|
bytesWritten += sentLength;
|
|
return (sentLength >= (NSInteger)size); // If we can't write all the data into the stream - bail-out early.
|
|
});
|
|
if (streamFailed) {
|
|
NSInteger code = 2145;
|
|
NSString *description = @"Error writing to stream.";
|
|
NSError *streamError = _outputStream.streamError;
|
|
NSError *error = streamError ? SRErrorWithCodeDescriptionUnderlyingError(code, description, streamError) : SRErrorWithCodeDescription(code, description);
|
|
[self _failWithError:error];
|
|
return;
|
|
}
|
|
|
|
_outputBufferOffset += bytesWritten;
|
|
|
|
if (_outputBufferOffset > SRDefaultBufferSize() && _outputBufferOffset > dataLength / 2) {
|
|
_outputBuffer = dispatch_data_create_subrange(_outputBuffer, _outputBufferOffset, dataLength - _outputBufferOffset);
|
|
_outputBufferOffset = 0;
|
|
}
|
|
}
|
|
|
|
if (_closeWhenFinishedWriting &&
|
|
(dispatch_data_get_size(_outputBuffer) - _outputBufferOffset) == 0 &&
|
|
(_inputStream.streamStatus != NSStreamStatusNotOpen &&
|
|
_inputStream.streamStatus != NSStreamStatusClosed) &&
|
|
!_sentClose) {
|
|
_sentClose = YES;
|
|
|
|
@synchronized(self) {
|
|
[_outputStream close];
|
|
[_inputStream close];
|
|
|
|
|
|
for (NSArray *runLoop in [_scheduledRunloops copy]) {
|
|
[self unscheduleFromRunLoop:[runLoop objectAtIndex:0] forMode:[runLoop objectAtIndex:1]];
|
|
}
|
|
}
|
|
|
|
if (!_failed) {
|
|
[self.delegateController performDelegateBlock:^(id<SRWebSocketDelegate> _Nullable delegate, SRDelegateAvailableMethods availableMethods) {
|
|
if (availableMethods.didCloseWithCode) {
|
|
[delegate webSocket:self didCloseWithCode:_closeCode reason:_closeReason wasClean:YES];
|
|
}
|
|
}];
|
|
}
|
|
|
|
[self _scheduleCleanup];
|
|
}
|
|
}
|
|
|
|
- (void)_addConsumerWithScanner:(stream_scanner)consumer callback:(data_callback)callback;
|
|
{
|
|
[self assertOnWorkQueue];
|
|
[self _addConsumerWithScanner:consumer callback:callback dataLength:0];
|
|
}
|
|
|
|
- (void)_addConsumerWithDataLength:(size_t)dataLength callback:(data_callback)callback readToCurrentFrame:(BOOL)readToCurrentFrame unmaskBytes:(BOOL)unmaskBytes;
|
|
{
|
|
[self assertOnWorkQueue];
|
|
assert(dataLength);
|
|
|
|
[_consumers addObject:[_consumerPool consumerWithScanner:nil handler:callback bytesNeeded:dataLength readToCurrentFrame:readToCurrentFrame unmaskBytes:unmaskBytes]];
|
|
[self _pumpScanner];
|
|
}
|
|
|
|
- (void)_addConsumerWithScanner:(stream_scanner)consumer callback:(data_callback)callback dataLength:(size_t)dataLength;
|
|
{
|
|
[self assertOnWorkQueue];
|
|
[_consumers addObject:[_consumerPool consumerWithScanner:consumer handler:callback bytesNeeded:dataLength readToCurrentFrame:NO unmaskBytes:NO]];
|
|
[self _pumpScanner];
|
|
}
|
|
|
|
|
|
- (void)_scheduleCleanup
|
|
{
|
|
@synchronized(self) {
|
|
if (_cleanupScheduled) {
|
|
return;
|
|
}
|
|
|
|
_cleanupScheduled = YES;
|
|
|
|
// _consumers retain SRWebSocket instance by block copy, if there are consumers here, clear them.
|
|
[_consumers removeAllObjects];
|
|
[_consumerPool clear];
|
|
|
|
// Cancel the timer which retains SRWebSocket instance.
|
|
// If we don't cancel the timer, the 'dealloc' method will be invoked only after the time (default: 60s) have come, which may cause memory increase.
|
|
dispatch_async(dispatch_get_main_queue(), ^(){
|
|
[NSObject cancelPreviousPerformRequestsWithTarget:self selector:@selector(_onTimeout) object:nil];
|
|
});
|
|
|
|
// Cleanup NSStream delegate's in the same RunLoop used by the streams themselves:
|
|
// This way we'll prevent race conditions between handleEvent and SRWebsocket's dealloc
|
|
NSTimer *timer = [NSTimer timerWithTimeInterval:(0.0f) target:self selector:@selector(_cleanupSelfReference:) userInfo:nil repeats:NO];
|
|
[[NSRunLoop SR_networkRunLoop] addTimer:timer forMode:NSDefaultRunLoopMode];
|
|
}
|
|
}
|
|
|
|
- (void)_cleanupSelfReference:(NSTimer *)timer
|
|
{
|
|
@synchronized(self) {
|
|
// Nuke NSStream delegate's
|
|
_inputStream.delegate = nil;
|
|
_outputStream.delegate = nil;
|
|
|
|
// Remove the streams, right now, from the networkRunLoop
|
|
[_inputStream close];
|
|
[_outputStream close];
|
|
}
|
|
|
|
// Cleanup selfRetain in the same GCD queue as usual
|
|
dispatch_async(_workQueue, ^{
|
|
_selfRetain = nil;
|
|
});
|
|
}
|
|
|
|
|
|
static const char CRLFCRLFBytes[] = {'\r', '\n', '\r', '\n'};
|
|
|
|
- (void)_readUntilHeaderCompleteWithCallback:(data_callback)dataHandler;
|
|
{
|
|
[self _readUntilBytes:CRLFCRLFBytes length:sizeof(CRLFCRLFBytes) callback:dataHandler];
|
|
}
|
|
|
|
- (void)_readUntilBytes:(const void *)bytes length:(size_t)length callback:(data_callback)dataHandler;
|
|
{
|
|
// TODO optimize so this can continue from where we last searched
|
|
stream_scanner consumer = ^size_t(NSData *data) {
|
|
__block size_t found_size = 0;
|
|
__block size_t match_count = 0;
|
|
|
|
size_t size = data.length;
|
|
const unsigned char *buffer = data.bytes;
|
|
for (size_t i = 0; i < size; i++ ) {
|
|
if (((const unsigned char *)buffer)[i] == ((const unsigned char *)bytes)[match_count]) {
|
|
match_count += 1;
|
|
if (match_count == length) {
|
|
found_size = i + 1;
|
|
break;
|
|
}
|
|
} else {
|
|
match_count = 0;
|
|
}
|
|
}
|
|
return found_size;
|
|
};
|
|
[self _addConsumerWithScanner:consumer callback:dataHandler];
|
|
}
|
|
|
|
|
|
// Returns true if did work
|
|
- (BOOL)_innerPumpScanner {
|
|
|
|
BOOL didWork = NO;
|
|
|
|
if (self.readyState >= SR_CLOSED) {
|
|
return didWork;
|
|
}
|
|
|
|
size_t readBufferSize = dispatch_data_get_size(_readBuffer);
|
|
|
|
if (!_consumers.count) {
|
|
return didWork;
|
|
}
|
|
|
|
size_t curSize = readBufferSize - _readBufferOffset;
|
|
if (!curSize) {
|
|
return didWork;
|
|
}
|
|
|
|
SRIOConsumer *consumer = [_consumers objectAtIndex:0];
|
|
|
|
size_t bytesNeeded = consumer.bytesNeeded;
|
|
|
|
size_t foundSize = 0;
|
|
if (consumer.consumer) {
|
|
NSData *subdata = (NSData *)dispatch_data_create_subrange(_readBuffer, _readBufferOffset, readBufferSize - _readBufferOffset);
|
|
foundSize = consumer.consumer(subdata);
|
|
} else {
|
|
assert(consumer.bytesNeeded);
|
|
if (curSize >= bytesNeeded) {
|
|
foundSize = bytesNeeded;
|
|
} else if (consumer.readToCurrentFrame) {
|
|
foundSize = curSize;
|
|
}
|
|
}
|
|
|
|
if (consumer.readToCurrentFrame || foundSize) {
|
|
dispatch_data_t slice = dispatch_data_create_subrange(_readBuffer, _readBufferOffset, foundSize);
|
|
|
|
_readBufferOffset += foundSize;
|
|
|
|
if (_readBufferOffset > SRDefaultBufferSize() && _readBufferOffset > readBufferSize / 2) {
|
|
_readBuffer = dispatch_data_create_subrange(_readBuffer, _readBufferOffset, readBufferSize - _readBufferOffset);
|
|
_readBufferOffset = 0;
|
|
}
|
|
|
|
if (consumer.unmaskBytes) {
|
|
__block NSMutableData *mutableSlice = [slice mutableCopy];
|
|
|
|
NSUInteger len = mutableSlice.length;
|
|
uint8_t *bytes = mutableSlice.mutableBytes;
|
|
|
|
for (NSUInteger i = 0; i < len; i++) {
|
|
bytes[i] = bytes[i] ^ _currentReadMaskKey[_currentReadMaskOffset % sizeof(_currentReadMaskKey)];
|
|
_currentReadMaskOffset += 1;
|
|
}
|
|
|
|
slice = dispatch_data_create(bytes, len, nil, ^{
|
|
mutableSlice = nil;
|
|
});
|
|
}
|
|
|
|
if (consumer.readToCurrentFrame) {
|
|
dispatch_data_apply(slice, ^bool(dispatch_data_t region, size_t offset, const void *buffer, size_t size) {
|
|
[_currentFrameData appendBytes:buffer length:size];
|
|
return true;
|
|
});
|
|
|
|
_readOpCount += 1;
|
|
|
|
if (_currentFrameOpcode == SROpCodeTextFrame) {
|
|
// Validate UTF8 stuff.
|
|
size_t currentDataSize = _currentFrameData.length;
|
|
if (_currentFrameOpcode == SROpCodeTextFrame && currentDataSize > 0) {
|
|
// TODO: Optimize the crap out of this. Don't really have to copy all the data each time
|
|
|
|
size_t scanSize = currentDataSize - _currentStringScanPosition;
|
|
|
|
NSData *scan_data = [_currentFrameData subdataWithRange:NSMakeRange(_currentStringScanPosition, scanSize)];
|
|
int32_t valid_utf8_size = validate_dispatch_data_partial_string(scan_data);
|
|
|
|
if (valid_utf8_size == -1) {
|
|
[self closeWithCode:SRStatusCodeInvalidUTF8 reason:@"Text frames must be valid UTF-8"];
|
|
dispatch_async(_workQueue, ^{
|
|
[self closeConnection];
|
|
});
|
|
return didWork;
|
|
} else {
|
|
_currentStringScanPosition += valid_utf8_size;
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
consumer.bytesNeeded -= foundSize;
|
|
|
|
if (consumer.bytesNeeded == 0) {
|
|
[_consumers removeObjectAtIndex:0];
|
|
consumer.handler(self, nil);
|
|
[_consumerPool returnConsumer:consumer];
|
|
didWork = YES;
|
|
}
|
|
} else if (foundSize) {
|
|
[_consumers removeObjectAtIndex:0];
|
|
consumer.handler(self, (NSData *)slice);
|
|
[_consumerPool returnConsumer:consumer];
|
|
didWork = YES;
|
|
}
|
|
}
|
|
return didWork;
|
|
}
|
|
|
|
-(void)_pumpScanner;
|
|
{
|
|
[self assertOnWorkQueue];
|
|
|
|
if (!_isPumping) {
|
|
_isPumping = YES;
|
|
} else {
|
|
return;
|
|
}
|
|
|
|
while ([self _innerPumpScanner]) {
|
|
|
|
}
|
|
|
|
_isPumping = NO;
|
|
}
|
|
|
|
//#define NOMASK
|
|
|
|
static const size_t SRFrameHeaderOverhead = 32;
|
|
|
|
- (void)_sendFrameWithOpcode:(SROpCode)opCode data:(NSData *)data
|
|
{
|
|
[self assertOnWorkQueue];
|
|
|
|
if (!data) {
|
|
return;
|
|
}
|
|
|
|
size_t payloadLength = data.length;
|
|
|
|
NSMutableData *frameData = [[NSMutableData alloc] initWithLength:payloadLength + SRFrameHeaderOverhead];
|
|
if (!frameData) {
|
|
[self closeWithCode:SRStatusCodeMessageTooBig reason:@"Message too big"];
|
|
return;
|
|
}
|
|
uint8_t *frameBuffer = (uint8_t *)frameData.mutableBytes;
|
|
|
|
// set fin
|
|
frameBuffer[0] = SRFinMask | opCode;
|
|
|
|
// set the mask and header
|
|
frameBuffer[1] |= SRMaskMask;
|
|
|
|
size_t frameBufferSize = 2;
|
|
|
|
if (payloadLength < 126) {
|
|
frameBuffer[1] |= payloadLength;
|
|
} else {
|
|
uint64_t declaredPayloadLength = 0;
|
|
size_t declaredPayloadLengthSize = 0;
|
|
|
|
if (payloadLength <= UINT16_MAX) {
|
|
frameBuffer[1] |= 126;
|
|
|
|
declaredPayloadLength = CFSwapInt16BigToHost((uint16_t)payloadLength);
|
|
declaredPayloadLengthSize = sizeof(uint16_t);
|
|
} else {
|
|
frameBuffer[1] |= 127;
|
|
|
|
declaredPayloadLength = CFSwapInt64BigToHost((uint64_t)payloadLength);
|
|
declaredPayloadLengthSize = sizeof(uint64_t);
|
|
}
|
|
|
|
memcpy((frameBuffer + frameBufferSize), &declaredPayloadLength, declaredPayloadLengthSize);
|
|
frameBufferSize += declaredPayloadLengthSize;
|
|
}
|
|
|
|
const uint8_t *unmaskedPayloadBuffer = (uint8_t *)data.bytes;
|
|
uint8_t *maskKey = frameBuffer + frameBufferSize;
|
|
|
|
size_t randomBytesSize = sizeof(uint32_t);
|
|
int result = SecRandomCopyBytes(kSecRandomDefault, randomBytesSize, maskKey);
|
|
if (result != 0) {
|
|
//TODO: (nlutsenko) Check if there was an error.
|
|
}
|
|
frameBufferSize += randomBytesSize;
|
|
|
|
// Copy and unmask the buffer
|
|
uint8_t *frameBufferPayloadPointer = frameBuffer + frameBufferSize;
|
|
|
|
memcpy(frameBufferPayloadPointer, unmaskedPayloadBuffer, payloadLength);
|
|
SRMaskBytesSIMD(frameBufferPayloadPointer, payloadLength, maskKey);
|
|
frameBufferSize += payloadLength;
|
|
|
|
assert(frameBufferSize <= frameData.length);
|
|
frameData.length = frameBufferSize;
|
|
|
|
[self _writeData:frameData];
|
|
}
|
|
|
|
- (void)stream:(NSStream *)aStream handleEvent:(NSStreamEvent)eventCode
|
|
{
|
|
__weak __typeof(self) wself = self;
|
|
|
|
if (_requestRequiresSSL && !_streamSecurityValidated &&
|
|
(eventCode == NSStreamEventHasBytesAvailable || eventCode == NSStreamEventHasSpaceAvailable)) {
|
|
SecTrustRef trust = (__bridge SecTrustRef)[aStream propertyForKey:(__bridge id)kCFStreamPropertySSLPeerTrust];
|
|
if (trust) {
|
|
_streamSecurityValidated = [_securityPolicy evaluateServerTrust:trust forDomain:_urlRequest.URL.host];
|
|
}
|
|
if (!_streamSecurityValidated) {
|
|
dispatch_async(_workQueue, ^{
|
|
NSError *error = SRErrorWithDomainCodeDescription(NSURLErrorDomain,
|
|
NSURLErrorClientCertificateRejected,
|
|
@"Invalid server certificate.");
|
|
[wself _failWithError:error];
|
|
});
|
|
return;
|
|
}
|
|
dispatch_async(_workQueue, ^{
|
|
[self didConnect];
|
|
});
|
|
}
|
|
dispatch_async(_workQueue, ^{
|
|
[wself safeHandleEvent:eventCode stream:aStream];
|
|
});
|
|
}
|
|
|
|
- (void)safeHandleEvent:(NSStreamEvent)eventCode stream:(NSStream *)aStream
|
|
{
|
|
switch (eventCode) {
|
|
case NSStreamEventOpenCompleted: {
|
|
SRDebugLog(@"NSStreamEventOpenCompleted %@", aStream);
|
|
if (self.readyState >= SR_CLOSING) {
|
|
return;
|
|
}
|
|
assert(_readBuffer);
|
|
|
|
if (!_requestRequiresSSL && self.readyState == SR_CONNECTING && aStream == _inputStream) {
|
|
[self didConnect];
|
|
}
|
|
|
|
[self _pumpWriting];
|
|
[self _pumpScanner];
|
|
|
|
break;
|
|
}
|
|
|
|
case NSStreamEventErrorOccurred: {
|
|
SRDebugLog(@"NSStreamEventErrorOccurred %@ %@", aStream, [[aStream streamError] copy]);
|
|
/// TODO specify error better!
|
|
[self _failWithError:aStream.streamError];
|
|
_readBufferOffset = 0;
|
|
_readBuffer = dispatch_data_empty;
|
|
break;
|
|
|
|
}
|
|
|
|
case NSStreamEventEndEncountered: {
|
|
[self _pumpScanner];
|
|
SRDebugLog(@"NSStreamEventEndEncountered %@", aStream);
|
|
if (aStream.streamError) {
|
|
[self _failWithError:aStream.streamError];
|
|
} else {
|
|
dispatch_async(_workQueue, ^{
|
|
if (self.readyState != SR_CLOSED) {
|
|
self.readyState = SR_CLOSED;
|
|
[self _scheduleCleanup];
|
|
}
|
|
|
|
if (!_sentClose && !_failed) {
|
|
_sentClose = YES;
|
|
// If we get closed in this state it's probably not clean because we should be sending this when we send messages
|
|
[self.delegateController performDelegateBlock:^(id<SRWebSocketDelegate> _Nullable delegate, SRDelegateAvailableMethods availableMethods) {
|
|
if (availableMethods.didCloseWithCode) {
|
|
[delegate webSocket:self
|
|
didCloseWithCode:SRStatusCodeGoingAway
|
|
reason:@"Stream end encountered"
|
|
wasClean:NO];
|
|
}
|
|
}];
|
|
}
|
|
});
|
|
}
|
|
|
|
break;
|
|
}
|
|
|
|
case NSStreamEventHasBytesAvailable: {
|
|
SRDebugLog(@"NSStreamEventHasBytesAvailable %@", aStream);
|
|
uint8_t buffer[SRDefaultBufferSize()];
|
|
|
|
while (_inputStream.hasBytesAvailable) {
|
|
NSInteger bytesRead = [_inputStream read:buffer maxLength:SRDefaultBufferSize()];
|
|
if (bytesRead > 0) {
|
|
dispatch_data_t data = dispatch_data_create(buffer, bytesRead, nil, DISPATCH_DATA_DESTRUCTOR_DEFAULT);
|
|
if (!data) {
|
|
NSError *error = SRErrorWithCodeDescription(SRStatusCodeMessageTooBig,
|
|
@"Unable to allocate memory to read from socket.");
|
|
[self _failWithError:error];
|
|
return;
|
|
}
|
|
_readBuffer = dispatch_data_create_concat(_readBuffer, data);
|
|
} else if (bytesRead == -1) {
|
|
[self _failWithError:_inputStream.streamError];
|
|
}
|
|
}
|
|
[self _pumpScanner];
|
|
break;
|
|
}
|
|
|
|
case NSStreamEventHasSpaceAvailable: {
|
|
SRDebugLog(@"NSStreamEventHasSpaceAvailable %@", aStream);
|
|
[self _pumpWriting];
|
|
break;
|
|
}
|
|
|
|
case NSStreamEventNone:
|
|
SRDebugLog(@"(default) %@", aStream);
|
|
break;
|
|
}
|
|
}
|
|
|
|
///--------------------------------------
|
|
#pragma mark - Delegate
|
|
///--------------------------------------
|
|
|
|
- (id<SRWebSocketDelegate> _Nullable)delegate
|
|
{
|
|
return self.delegateController.delegate;
|
|
}
|
|
|
|
- (void)setDelegate:(id<SRWebSocketDelegate> _Nullable)delegate
|
|
{
|
|
self.delegateController.delegate = delegate;
|
|
}
|
|
|
|
- (void)setDelegateDispatchQueue:(dispatch_queue_t _Nullable)queue
|
|
{
|
|
self.delegateController.dispatchQueue = queue;
|
|
}
|
|
|
|
- (dispatch_queue_t _Nullable)delegateDispatchQueue
|
|
{
|
|
return self.delegateController.dispatchQueue;
|
|
}
|
|
|
|
- (void)setDelegateOperationQueue:(NSOperationQueue *_Nullable)queue
|
|
{
|
|
self.delegateController.operationQueue = queue;
|
|
}
|
|
|
|
- (NSOperationQueue *_Nullable)delegateOperationQueue
|
|
{
|
|
return self.delegateController.operationQueue;
|
|
}
|
|
|
|
@end
|
|
|
|
#ifdef HAS_ICU
|
|
|
|
static inline int32_t validate_dispatch_data_partial_string(NSData *data) {
|
|
if ([data length] > INT32_MAX) {
|
|
// INT32_MAX is the limit so long as this Framework is using 32 bit ints everywhere.
|
|
return -1;
|
|
}
|
|
|
|
int32_t size = (int32_t)[data length];
|
|
|
|
const void * contents = [data bytes];
|
|
const uint8_t *str = (const uint8_t *)contents;
|
|
|
|
UChar32 codepoint = 1;
|
|
int32_t offset = 0;
|
|
int32_t lastOffset = 0;
|
|
while(offset < size && codepoint > 0) {
|
|
lastOffset = offset;
|
|
U8_NEXT(str, offset, size, codepoint);
|
|
}
|
|
|
|
if (codepoint == -1) {
|
|
// Check to see if the last byte is valid or whether it was just continuing
|
|
if (!U8_IS_LEAD(str[lastOffset]) || U8_COUNT_TRAIL_BYTES(str[lastOffset]) + lastOffset < (int32_t)size) {
|
|
|
|
size = -1;
|
|
} else {
|
|
uint8_t leadByte = str[lastOffset];
|
|
U8_MASK_LEAD_BYTE(leadByte, U8_COUNT_TRAIL_BYTES(leadByte));
|
|
|
|
for (int i = lastOffset + 1; i < offset; i++) {
|
|
if (U8_IS_SINGLE(str[i]) || U8_IS_LEAD(str[i]) || !U8_IS_TRAIL(str[i])) {
|
|
size = -1;
|
|
}
|
|
}
|
|
|
|
if (size != -1) {
|
|
size = lastOffset;
|
|
}
|
|
}
|
|
}
|
|
|
|
if (size != -1 && ![[NSString alloc] initWithBytesNoCopy:(char *)[data bytes] length:size encoding:NSUTF8StringEncoding freeWhenDone:NO]) {
|
|
size = -1;
|
|
}
|
|
|
|
return size;
|
|
}
|
|
|
|
#else
|
|
|
|
// This is a hack, and probably not optimal
|
|
static inline int32_t validate_dispatch_data_partial_string(NSData *data) {
|
|
static const int maxCodepointSize = 3;
|
|
|
|
for (int i = 0; i < maxCodepointSize; i++) {
|
|
NSString *str = [[NSString alloc] initWithBytesNoCopy:(char *)data.bytes length:data.length - i encoding:NSUTF8StringEncoding freeWhenDone:NO];
|
|
if (str) {
|
|
return (int32_t)data.length - i;
|
|
}
|
|
}
|
|
|
|
return -1;
|
|
}
|
|
|
|
#endif
|