Skip to content

Commit

Permalink
Merge pull request #159 from vapor/beta-1
Browse files Browse the repository at this point in the history 
beta 1
  • Loading branch information
@tanner0101
tanner0101 authored Feb 9, 2018
2 parents 063c356 + 8250a38 commit 70b00ab
Show file tree
Hide file tree
Showing 6 changed files with 38 additions and 19 deletions.
3 changes: 2 additions & 1 deletion Sources/TCP/Socket/TCPClient.swift
View file
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,8 @@ extension TCPSocket {
// http://www.madore.org/~david/computers/connect-intr.html
case EINPROGRESS:
if !isNonBlocking {
fatalError("EINPROGRESS on a blocking socket")
ERROR("EINPROGRESS on a blocking socket")
return
}
default: throw TCPError.posix(errno, identifier: "connect")
}
Expand Down
31 changes: 20 additions & 11 deletions Sources/TCP/Streams/TCPSocketSink.swift
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,11 +60,13 @@ public final class TCPSocketSink: Async.InputStream {
switch event {
case .next(let input, let ready):
guard inputBuffer == nil else {
fatalError("SocketSink upstream is illegally overproducing input buffers.")
ERROR("SocketSink upstream is illegally overproducing input buffers.")
return
}
inputBuffer = input
guard currentReadyPromise == nil else {
fatalError("SocketSink currentReadyPromise illegally not nil during input.")
ERROR("SocketSink currentReadyPromise illegally not nil during input.")
return
}
currentReadyPromise = ready
resumeIfSuspended()
Expand All @@ -81,7 +83,8 @@ public final class TCPSocketSink: Async.InputStream {
return
}
guard let writeSource = self.writeSource else {
fatalError("SocketSink writeSource illegally nil during close.")
ERROR("SocketSink writeSource illegally nil during close.")
return
}
writeSource.cancel()
socket.close()
Expand All @@ -93,7 +96,8 @@ public final class TCPSocketSink: Async.InputStream {
private func writeData(ready: Promise<Void>) {
do {
guard let buffer = self.inputBuffer else {
fatalError("Unexpected nil SocketSink inputBuffer during writeData")
ERROR("Unexpected nil SocketSink inputBuffer during writeData")
return
}

let write = try socket.write(from: buffer)
Expand All @@ -102,7 +106,7 @@ public final class TCPSocketSink: Async.InputStream {
switch count {
case buffer.count:
self.inputBuffer = nil
ready.complete()
ready.complete(onNextTick: eventLoop)
default:
inputBuffer = ByteBuffer(
start: buffer.baseAddress?.advanced(by: count),
Expand All @@ -113,13 +117,14 @@ public final class TCPSocketSink: Async.InputStream {
case .wouldBlock:
resumeIfSuspended()
guard currentReadyPromise == nil else {
fatalError("SocketSink currentReadyPromise illegally not nil during wouldBlock.")
ERROR("SocketSink currentReadyPromise illegally not nil during wouldBlock.")
return
}
currentReadyPromise = ready
}
} catch {
self.error(error)
ready.complete()
ready.complete(onNextTick: eventLoop)
}
}

Expand All @@ -136,7 +141,8 @@ public final class TCPSocketSink: Async.InputStream {
excessSignalCount = excessSignalCount &+ 1
if excessSignalCount >= maxExcessSignalCount {
guard let writeSource = self.writeSource else {
fatalError("SocketSink writeSource illegally nil during signal.")
ERROR("SocketSink writeSource illegally nil during signal.")
return
}
writeSource.suspend()
sourceIsSuspended = true
Expand All @@ -145,7 +151,8 @@ public final class TCPSocketSink: Async.InputStream {
}

guard let ready = currentReadyPromise else {
fatalError("SocketSink currentReadyPromise illegaly nil during signal.")
ERROR("SocketSink currentReadyPromise illegaly nil during signal.")
return
}
currentReadyPromise = nil
writeData(ready: ready)
Expand All @@ -157,7 +164,8 @@ public final class TCPSocketSink: Async.InputStream {
}

guard let writeSource = self.writeSource else {
fatalError("SocketSink writeSource illegally nil during resumeIfSuspended.")
ERROR("SocketSink writeSource illegally nil during resumeIfSuspended.")
return
}
sourceIsSuspended = false
// start listening for ready notifications
Expand All @@ -177,7 +185,8 @@ extension TCPSocket {
@available(*, deprecated)
public func sink(on eventLoop: Worker) -> TCPSocketSink {
return .init(socket: self, on: eventLoop) { _, error in
fatalError("Uncaught error in TCPSocketSink: \(error).")
ERROR("Uncaught error in TCPSocketSink: \(error).")
return
}
}
}
12 changes: 8 additions & 4 deletions Sources/TCP/Streams/TCPSocketSource.swift
View file
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,8 @@ public final class TCPSocketSource: Async.OutputStream {
return
}
guard let readSource = self.readSource else {
fatalError("SocketSource readSource illegally nil during close.")
ERROR("SocketSource readSource illegally nil during close.")
return
}
readSource.cancel()
socket.close()
Expand All @@ -84,7 +85,8 @@ public final class TCPSocketSource: Async.OutputStream {
/// as indicated by a read source.
private func readData() {
guard let downstream = self.downstream else {
fatalError("Unexpected nil downstream on SocketSource during readData.")
ERROR("Unexpected nil downstream on SocketSource during readData.")
return
}
do {
let read = try socket.read(into: buffer)
Expand Down Expand Up @@ -141,7 +143,8 @@ public final class TCPSocketSource: Async.OutputStream {
excessSignalCount = excessSignalCount &+ 1
if excessSignalCount >= maxExcessSignalCount {
guard let readSource = self.readSource else {
fatalError("SocketSource readSource illegally nil during signal.")
ERROR("SocketSource readSource illegally nil during signal.")
return
}
readSource.suspend()
sourceIsSuspended = true
Expand All @@ -161,7 +164,8 @@ public final class TCPSocketSource: Async.OutputStream {
}

guard let readSource = self.readSource else {
fatalError("SocketSource readSource illegally nil on resumeIfSuspended.")
ERROR("SocketSource readSource illegally nil on resumeIfSuspended.")
return
}
sourceIsSuspended = false
readSource.resume()
Expand Down
3 changes: 2 additions & 1 deletion Sources/TCP/Streams/TCPSocketStream.swift
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,8 @@ extension TCPSocket {
@available(*, deprecated)
public func stream(bufferSize: Int = 4096, on worker: Worker) -> TCPSocketStream {
return TCPSocketStream(socket: self, bufferSize: bufferSize, on: worker) { _, error in
fatalError("Uncaught error in TCPSocketStream: \(error).")
ERROR("Uncaught error in TCPSocketStream: \(error).")
return
}
}
}
4 changes: 3 additions & 1 deletion Sources/TCP/Utilties/TCPError.swift
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,4 +62,6 @@ public struct TCPError: Traceable, Debuggable, Helpable, Swift.Error, Encodable
}
}


func ERROR(_ message: String, file: StaticString = #file, line: Int = #line) {
print("[TCP] \(message) [\(file):\(line)]")
}
4 changes: 3 additions & 1 deletion Sources/TCP/Utilties/TCPSocket+Data.swift
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,9 @@ extension TCPSocket {

switch read {
case .success(let count): data.removeLast(data.count &- count)
case .wouldBlock: fatalError()
case .wouldBlock:
ERROR("TCPSocket blocked during read Data.")
return data
}

return data
Expand Down

0 comments on commit 70b00ab

Please sign in to comment.