asyncnet.nim 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822
  1. #
  2. #
  3. # Nim's Runtime Library
  4. # (c) Copyright 2017 Dominik Picheta
  5. #
  6. # See the file "copying.txt", included in this
  7. # distribution, for details about the copyright.
  8. #
  9. ## This module implements a high-level asynchronous sockets API based on the
  10. ## asynchronous dispatcher defined in the ``asyncdispatch`` module.
  11. ##
  12. ## Asynchronous IO in Nim
  13. ## ======================
  14. ##
  15. ## Async IO in Nim consists of multiple layers (from highest to lowest):
  16. ##
  17. ## * ``asyncnet`` module
  18. ##
  19. ## * Async await
  20. ##
  21. ## * ``asyncdispatch`` module (event loop)
  22. ##
  23. ## * ``selectors`` module
  24. ##
  25. ## Each builds on top of the layers below it. The selectors module is an
  26. ## abstraction for the various system ``select()`` mechanisms such as epoll or
  27. ## kqueue. If you wish you can use it directly, and some people have done so
  28. ## `successfully <http://goran.krampe.se/2014/10/25/nim-socketserver/>`_.
  29. ## But you must be aware that on Windows it only supports
  30. ## ``select()``.
  31. ##
  32. ## The async dispatcher implements the proactor pattern and also has an
  33. ## implementation of IOCP. It implements the proactor pattern for other
  34. ## OS' via the selectors module. Futures are also implemented here, and
  35. ## indeed all the procedures return a future.
  36. ##
  37. ## The final layer is the async await transformation. This allows you to
  38. ## write asynchronous code in a synchronous style and works similar to
  39. ## C#'s await. The transformation works by converting any async procedures
  40. ## into an iterator.
  41. ##
  42. ## This is all single threaded, fully non-blocking and does give you a
  43. ## lot of control. In theory you should be able to work with any of these
  44. ## layers interchangeably (as long as you only care about non-Windows
  45. ## platforms).
  46. ##
  47. ## For most applications using ``asyncnet`` is the way to go as it builds
  48. ## over all the layers, providing some extra features such as buffering.
  49. ##
  50. ## SSL
  51. ## ===
  52. ##
  53. ## SSL can be enabled by compiling with the ``-d:ssl`` flag.
  54. ##
  55. ## You must create a new SSL context with the ``newContext`` function defined
  56. ## in the ``net`` module. You may then call ``wrapSocket`` on your socket using
  57. ## the newly created SSL context to get an SSL socket.
  58. ##
  59. ## Examples
  60. ## ========
  61. ##
  62. ## Chat server
  63. ## -----------
  64. ##
  65. ## The following example demonstrates a simple chat server.
  66. ##
  67. ## .. code-block::nim
  68. ##
  69. ## import asyncnet, asyncdispatch
  70. ##
  71. ## var clients {.threadvar.}: seq[AsyncSocket]
  72. ##
  73. ## proc processClient(client: AsyncSocket) {.async.} =
  74. ## while true:
  75. ## let line = await client.recvLine()
  76. ## if line.len == 0: break
  77. ## for c in clients:
  78. ## await c.send(line & "\c\L")
  79. ##
  80. ## proc serve() {.async.} =
  81. ## clients = @[]
  82. ## var server = newAsyncSocket()
  83. ## server.setSockOpt(OptReuseAddr, true)
  84. ## server.bindAddr(Port(12345))
  85. ## server.listen()
  86. ##
  87. ## while true:
  88. ## let client = await server.accept()
  89. ## clients.add client
  90. ##
  91. ## asyncCheck processClient(client)
  92. ##
  93. ## asyncCheck serve()
  94. ## runForever()
  95. ##
  96. import asyncdispatch
  97. import nativesockets
  98. import net
  99. import os
  100. export SOBool
  101. # TODO: Remove duplication introduced by PR #4683.
  102. const defineSsl = defined(ssl) or defined(nimdoc)
  103. when defineSsl:
  104. import openssl
  105. type
  106. # TODO: I would prefer to just do:
  107. # AsyncSocket* {.borrow: `.`.} = distinct Socket. But that doesn't work.
  108. AsyncSocketDesc = object
  109. fd: SocketHandle
  110. closed: bool ## determines whether this socket has been closed
  111. isBuffered: bool ## determines whether this socket is buffered.
  112. buffer: array[0..BufferSize, char]
  113. currPos: int # current index in buffer
  114. bufLen: int # current length of buffer
  115. isSsl: bool
  116. when defineSsl:
  117. sslHandle: SslPtr
  118. sslContext: SslContext
  119. bioIn: BIO
  120. bioOut: BIO
  121. domain: Domain
  122. sockType: SockType
  123. protocol: Protocol
  124. AsyncSocket* = ref AsyncSocketDesc
  125. proc newAsyncSocket*(fd: AsyncFD, domain: Domain = AF_INET,
  126. sockType: SockType = SOCK_STREAM,
  127. protocol: Protocol = IPPROTO_TCP, buffered = true): owned(AsyncSocket) =
  128. ## Creates a new ``AsyncSocket`` based on the supplied params.
  129. ##
  130. ## The supplied ``fd``'s non-blocking state will be enabled implicitly.
  131. ##
  132. ## **Note**: This procedure will **NOT** register ``fd`` with the global
  133. ## async dispatcher. You need to do this manually. If you have used
  134. ## ``newAsyncNativeSocket`` to create ``fd`` then it's already registered.
  135. assert fd != osInvalidSocket.AsyncFD
  136. new(result)
  137. result.fd = fd.SocketHandle
  138. fd.SocketHandle.setBlocking(false)
  139. result.isBuffered = buffered
  140. result.domain = domain
  141. result.sockType = sockType
  142. result.protocol = protocol
  143. if buffered:
  144. result.currPos = 0
  145. proc newAsyncSocket*(domain: Domain = AF_INET, sockType: SockType = SOCK_STREAM,
  146. protocol: Protocol = IPPROTO_TCP, buffered = true): owned(AsyncSocket) =
  147. ## Creates a new asynchronous socket.
  148. ##
  149. ## This procedure will also create a brand new file descriptor for
  150. ## this socket.
  151. let fd = createAsyncNativeSocket(domain, sockType, protocol)
  152. if fd.SocketHandle == osInvalidSocket:
  153. raiseOSError(osLastError())
  154. result = newAsyncSocket(fd, domain, sockType, protocol, buffered)
  155. proc getLocalAddr*(socket: AsyncSocket): (string, Port) =
  156. ## Get the socket's local address and port number.
  157. ##
  158. ## This is high-level interface for `getsockname`:idx:.
  159. getLocalAddr(socket.fd, socket.domain)
  160. proc getPeerAddr*(socket: AsyncSocket): (string, Port) =
  161. ## Get the socket's peer address and port number.
  162. ##
  163. ## This is high-level interface for `getpeername`:idx:.
  164. getPeerAddr(socket.fd, socket.domain)
  165. proc newAsyncSocket*(domain, sockType, protocol: cint,
  166. buffered = true): owned(AsyncSocket) =
  167. ## Creates a new asynchronous socket.
  168. ##
  169. ## This procedure will also create a brand new file descriptor for
  170. ## this socket.
  171. let fd = createAsyncNativeSocket(domain, sockType, protocol)
  172. if fd.SocketHandle == osInvalidSocket:
  173. raiseOSError(osLastError())
  174. result = newAsyncSocket(fd, Domain(domain), SockType(sockType),
  175. Protocol(protocol), buffered)
  176. when defineSsl:
  177. proc getSslError(handle: SslPtr, err: cint): cint =
  178. assert err < 0
  179. var ret = SSL_get_error(handle, err.cint)
  180. case ret
  181. of SSL_ERROR_ZERO_RETURN:
  182. raiseSSLError("TLS/SSL connection failed to initiate, socket closed prematurely.")
  183. of SSL_ERROR_WANT_CONNECT, SSL_ERROR_WANT_ACCEPT:
  184. return ret
  185. of SSL_ERROR_WANT_WRITE, SSL_ERROR_WANT_READ:
  186. return ret
  187. of SSL_ERROR_WANT_X509_LOOKUP:
  188. raiseSSLError("Function for x509 lookup has been called.")
  189. of SSL_ERROR_SYSCALL, SSL_ERROR_SSL:
  190. raiseSSLError()
  191. else: raiseSSLError("Unknown Error")
  192. proc sendPendingSslData(socket: AsyncSocket,
  193. flags: set[SocketFlag]) {.async.} =
  194. let len = bioCtrlPending(socket.bioOut)
  195. if len > 0:
  196. var data = newString(len)
  197. let read = bioRead(socket.bioOut, addr data[0], len)
  198. assert read != 0
  199. if read < 0:
  200. raiseSSLError()
  201. data.setLen(read)
  202. await socket.fd.AsyncFD.send(data, flags)
  203. proc appeaseSsl(socket: AsyncSocket, flags: set[SocketFlag],
  204. sslError: cint): owned(Future[bool]) {.async.} =
  205. ## Returns ``true`` if ``socket`` is still connected, otherwise ``false``.
  206. result = true
  207. case sslError
  208. of SSL_ERROR_WANT_WRITE:
  209. await sendPendingSslData(socket, flags)
  210. of SSL_ERROR_WANT_READ:
  211. var data = await recv(socket.fd.AsyncFD, BufferSize, flags)
  212. let length = len(data)
  213. if length > 0:
  214. let ret = bioWrite(socket.bioIn, addr data[0], length.cint)
  215. if ret < 0:
  216. raiseSSLError()
  217. elif length == 0:
  218. # connection not properly closed by remote side or connection dropped
  219. SSL_set_shutdown(socket.sslHandle, SSL_RECEIVED_SHUTDOWN)
  220. result = false
  221. else:
  222. raiseSSLError("Cannot appease SSL.")
  223. template sslLoop(socket: AsyncSocket, flags: set[SocketFlag],
  224. op: untyped) =
  225. var opResult {.inject.} = -1.cint
  226. while opResult < 0:
  227. # Call the desired operation.
  228. opResult = op
  229. # Bit hackish here.
  230. # TODO: Introduce an async template transformation pragma?
  231. # Send any remaining pending SSL data.
  232. yield sendPendingSslData(socket, flags)
  233. # If the operation failed, try to see if SSL has some data to read
  234. # or write.
  235. if opResult < 0:
  236. let err = getSslError(socket.sslHandle, opResult.cint)
  237. let fut = appeaseSsl(socket, flags, err.cint)
  238. yield fut
  239. if not fut.read():
  240. # Socket disconnected.
  241. if SocketFlag.SafeDisconn in flags:
  242. opResult = 0.cint
  243. break
  244. else:
  245. raiseSSLError("Socket has been disconnected")
  246. proc dial*(address: string, port: Port, protocol = IPPROTO_TCP,
  247. buffered = true): owned(Future[AsyncSocket]) {.async.} =
  248. ## Establishes connection to the specified ``address``:``port`` pair via the
  249. ## specified protocol. The procedure iterates through possible
  250. ## resolutions of the ``address`` until it succeeds, meaning that it
  251. ## seamlessly works with both IPv4 and IPv6.
  252. ## Returns AsyncSocket ready to send or receive data.
  253. let asyncFd = await asyncdispatch.dial(address, port, protocol)
  254. let sockType = protocol.toSockType()
  255. let domain = getSockDomain(asyncFd.SocketHandle)
  256. result = newAsyncSocket(asyncFd, domain, sockType, protocol, buffered)
  257. proc connect*(socket: AsyncSocket, address: string, port: Port) {.async.} =
  258. ## Connects ``socket`` to server at ``address:port``.
  259. ##
  260. ## Returns a ``Future`` which will complete when the connection succeeds
  261. ## or an error occurs.
  262. await connect(socket.fd.AsyncFD, address, port, socket.domain)
  263. if socket.isSsl:
  264. when defineSsl:
  265. if not isIpAddress(address):
  266. # Set the SNI address for this connection. This call can fail if
  267. # we're not using TLSv1+.
  268. discard SSL_set_tlsext_host_name(socket.sslHandle, address)
  269. let flags = {SocketFlag.SafeDisconn}
  270. sslSetConnectState(socket.sslHandle)
  271. sslLoop(socket, flags, sslDoHandshake(socket.sslHandle))
  272. template readInto(buf: pointer, size: int, socket: AsyncSocket,
  273. flags: set[SocketFlag]): int =
  274. ## Reads **up to** ``size`` bytes from ``socket`` into ``buf``. Note that
  275. ## this is a template and not a proc.
  276. assert(not socket.closed, "Cannot `recv` on a closed socket")
  277. var res = 0
  278. if socket.isSsl:
  279. when defineSsl:
  280. # SSL mode.
  281. sslLoop(socket, flags,
  282. sslRead(socket.sslHandle, cast[cstring](buf), size.cint))
  283. res = opResult
  284. else:
  285. var recvIntoFut = asyncdispatch.recvInto(socket.fd.AsyncFD, buf, size, flags)
  286. yield recvIntoFut
  287. # Not in SSL mode.
  288. res = recvIntoFut.read()
  289. res
  290. template readIntoBuf(socket: AsyncSocket,
  291. flags: set[SocketFlag]): int =
  292. var size = readInto(addr socket.buffer[0], BufferSize, socket, flags)
  293. socket.currPos = 0
  294. socket.bufLen = size
  295. size
  296. proc recvInto*(socket: AsyncSocket, buf: pointer, size: int,
  297. flags = {SocketFlag.SafeDisconn}): owned(Future[int]) {.async.} =
  298. ## Reads **up to** ``size`` bytes from ``socket`` into ``buf``.
  299. ##
  300. ## For buffered sockets this function will attempt to read all the requested
  301. ## data. It will read this data in ``BufferSize`` chunks.
  302. ##
  303. ## For unbuffered sockets this function makes no effort to read
  304. ## all the data requested. It will return as much data as the operating system
  305. ## gives it.
  306. ##
  307. ## If socket is disconnected during the
  308. ## recv operation then the future may complete with only a part of the
  309. ## requested data.
  310. ##
  311. ## If socket is disconnected and no data is available
  312. ## to be read then the future will complete with a value of ``0``.
  313. if socket.isBuffered:
  314. let originalBufPos = socket.currPos
  315. if socket.bufLen == 0:
  316. let res = socket.readIntoBuf(flags - {SocketFlag.Peek})
  317. if res == 0:
  318. return 0
  319. var read = 0
  320. var cbuf = cast[cstring](buf)
  321. while read < size:
  322. if socket.currPos >= socket.bufLen:
  323. if SocketFlag.Peek in flags:
  324. # We don't want to get another buffer if we're peeking.
  325. break
  326. let res = socket.readIntoBuf(flags - {SocketFlag.Peek})
  327. if res == 0:
  328. break
  329. let chunk = min(socket.bufLen-socket.currPos, size-read)
  330. copyMem(addr(cbuf[read]), addr(socket.buffer[socket.currPos]), chunk)
  331. read.inc(chunk)
  332. socket.currPos.inc(chunk)
  333. if SocketFlag.Peek in flags:
  334. # Restore old buffer cursor position.
  335. socket.currPos = originalBufPos
  336. result = read
  337. else:
  338. result = readInto(buf, size, socket, flags)
  339. proc recv*(socket: AsyncSocket, size: int,
  340. flags = {SocketFlag.SafeDisconn}): owned(Future[string]) {.async.} =
  341. ## Reads **up to** ``size`` bytes from ``socket``.
  342. ##
  343. ## For buffered sockets this function will attempt to read all the requested
  344. ## data. It will read this data in ``BufferSize`` chunks.
  345. ##
  346. ## For unbuffered sockets this function makes no effort to read
  347. ## all the data requested. It will return as much data as the operating system
  348. ## gives it.
  349. ##
  350. ## If socket is disconnected during the
  351. ## recv operation then the future may complete with only a part of the
  352. ## requested data.
  353. ##
  354. ## If socket is disconnected and no data is available
  355. ## to be read then the future will complete with a value of ``""``.
  356. if socket.isBuffered:
  357. result = newString(size)
  358. shallow(result)
  359. let originalBufPos = socket.currPos
  360. if socket.bufLen == 0:
  361. let res = socket.readIntoBuf(flags - {SocketFlag.Peek})
  362. if res == 0:
  363. result.setLen(0)
  364. return
  365. var read = 0
  366. while read < size:
  367. if socket.currPos >= socket.bufLen:
  368. if SocketFlag.Peek in flags:
  369. # We don't want to get another buffer if we're peeking.
  370. break
  371. let res = socket.readIntoBuf(flags - {SocketFlag.Peek})
  372. if res == 0:
  373. break
  374. let chunk = min(socket.bufLen-socket.currPos, size-read)
  375. copyMem(addr(result[read]), addr(socket.buffer[socket.currPos]), chunk)
  376. read.inc(chunk)
  377. socket.currPos.inc(chunk)
  378. if SocketFlag.Peek in flags:
  379. # Restore old buffer cursor position.
  380. socket.currPos = originalBufPos
  381. result.setLen(read)
  382. else:
  383. result = newString(size)
  384. let read = readInto(addr result[0], size, socket, flags)
  385. result.setLen(read)
  386. proc send*(socket: AsyncSocket, buf: pointer, size: int,
  387. flags = {SocketFlag.SafeDisconn}) {.async.} =
  388. ## Sends ``size`` bytes from ``buf`` to ``socket``. The returned future will complete once all
  389. ## data has been sent.
  390. assert socket != nil
  391. assert(not socket.closed, "Cannot `send` on a closed socket")
  392. if socket.isSsl:
  393. when defineSsl:
  394. sslLoop(socket, flags,
  395. sslWrite(socket.sslHandle, cast[cstring](buf), size.cint))
  396. await sendPendingSslData(socket, flags)
  397. else:
  398. await send(socket.fd.AsyncFD, buf, size, flags)
  399. proc send*(socket: AsyncSocket, data: string,
  400. flags = {SocketFlag.SafeDisconn}) {.async.} =
  401. ## Sends ``data`` to ``socket``. The returned future will complete once all
  402. ## data has been sent.
  403. assert socket != nil
  404. if socket.isSsl:
  405. when defineSsl:
  406. var copy = data
  407. sslLoop(socket, flags,
  408. sslWrite(socket.sslHandle, addr copy[0], copy.len.cint))
  409. await sendPendingSslData(socket, flags)
  410. else:
  411. await send(socket.fd.AsyncFD, data, flags)
  412. proc acceptAddr*(socket: AsyncSocket, flags = {SocketFlag.SafeDisconn}):
  413. owned(Future[tuple[address: string, client: AsyncSocket]]) =
  414. ## Accepts a new connection. Returns a future containing the client socket
  415. ## corresponding to that connection and the remote address of the client.
  416. ## The future will complete when the connection is successfully accepted.
  417. var retFuture = newFuture[tuple[address: string, client: AsyncSocket]]("asyncnet.acceptAddr")
  418. var fut = acceptAddr(socket.fd.AsyncFD, flags)
  419. fut.callback =
  420. proc (future: Future[tuple[address: string, client: AsyncFD]]) =
  421. assert future.finished
  422. if future.failed:
  423. retFuture.fail(future.readError)
  424. else:
  425. let resultTup = (future.read.address,
  426. newAsyncSocket(future.read.client, socket.domain,
  427. socket.sockType, socket.protocol, socket.isBuffered))
  428. retFuture.complete(resultTup)
  429. return retFuture
  430. proc accept*(socket: AsyncSocket,
  431. flags = {SocketFlag.SafeDisconn}): owned(Future[AsyncSocket]) =
  432. ## Accepts a new connection. Returns a future containing the client socket
  433. ## corresponding to that connection.
  434. ## The future will complete when the connection is successfully accepted.
  435. var retFut = newFuture[AsyncSocket]("asyncnet.accept")
  436. var fut = acceptAddr(socket, flags)
  437. fut.callback =
  438. proc (future: Future[tuple[address: string, client: AsyncSocket]]) =
  439. assert future.finished
  440. if future.failed:
  441. retFut.fail(future.readError)
  442. else:
  443. retFut.complete(future.read.client)
  444. return retFut
  445. proc recvLineInto*(socket: AsyncSocket, resString: FutureVar[string],
  446. flags = {SocketFlag.SafeDisconn}, maxLength = MaxLineLength) {.async.} =
  447. ## Reads a line of data from ``socket`` into ``resString``.
  448. ##
  449. ## If a full line is read ``\r\L`` is not
  450. ## added to ``line``, however if solely ``\r\L`` is read then ``line``
  451. ## will be set to it.
  452. ##
  453. ## If the socket is disconnected, ``line`` will be set to ``""``.
  454. ##
  455. ## If the socket is disconnected in the middle of a line (before ``\r\L``
  456. ## is read) then line will be set to ``""``.
  457. ## The partial line **will be lost**.
  458. ##
  459. ## The ``maxLength`` parameter determines the maximum amount of characters
  460. ## that can be read. ``resString`` will be truncated after that.
  461. ##
  462. ## **Warning**: The ``Peek`` flag is not yet implemented.
  463. ##
  464. ## **Warning**: ``recvLineInto`` on unbuffered sockets assumes that the
  465. ## protocol uses ``\r\L`` to delimit a new line.
  466. assert SocketFlag.Peek notin flags ## TODO:
  467. result = newFuture[void]("asyncnet.recvLineInto")
  468. # TODO: Make the async transformation check for FutureVar params and complete
  469. # them when the result future is completed.
  470. # Can we replace the result future with the FutureVar?
  471. template addNLIfEmpty(): untyped =
  472. if resString.mget.len == 0:
  473. resString.mget.add("\c\L")
  474. if socket.isBuffered:
  475. if socket.bufLen == 0:
  476. let res = socket.readIntoBuf(flags)
  477. if res == 0:
  478. resString.complete()
  479. return
  480. var lastR = false
  481. while true:
  482. if socket.currPos >= socket.bufLen:
  483. let res = socket.readIntoBuf(flags)
  484. if res == 0:
  485. resString.mget.setLen(0)
  486. resString.complete()
  487. return
  488. case socket.buffer[socket.currPos]
  489. of '\r':
  490. lastR = true
  491. addNLIfEmpty()
  492. of '\L':
  493. addNLIfEmpty()
  494. socket.currPos.inc()
  495. resString.complete()
  496. return
  497. else:
  498. if lastR:
  499. socket.currPos.inc()
  500. resString.complete()
  501. return
  502. else:
  503. resString.mget.add socket.buffer[socket.currPos]
  504. socket.currPos.inc()
  505. # Verify that this isn't a DOS attack: #3847.
  506. if resString.mget.len > maxLength: break
  507. else:
  508. var c = ""
  509. while true:
  510. c = await recv(socket, 1, flags)
  511. if c.len == 0:
  512. resString.mget.setLen(0)
  513. resString.complete()
  514. return
  515. if c == "\r":
  516. c = await recv(socket, 1, flags) # Skip \L
  517. assert c == "\L"
  518. addNLIfEmpty()
  519. resString.complete()
  520. return
  521. elif c == "\L":
  522. addNLIfEmpty()
  523. resString.complete()
  524. return
  525. resString.mget.add c
  526. # Verify that this isn't a DOS attack: #3847.
  527. if resString.mget.len > maxLength: break
  528. resString.complete()
  529. proc recvLine*(socket: AsyncSocket,
  530. flags = {SocketFlag.SafeDisconn},
  531. maxLength = MaxLineLength): owned(Future[string]) {.async.} =
  532. ## Reads a line of data from ``socket``. Returned future will complete once
  533. ## a full line is read or an error occurs.
  534. ##
  535. ## If a full line is read ``\r\L`` is not
  536. ## added to ``line``, however if solely ``\r\L`` is read then ``line``
  537. ## will be set to it.
  538. ##
  539. ## If the socket is disconnected, ``line`` will be set to ``""``.
  540. ##
  541. ## If the socket is disconnected in the middle of a line (before ``\r\L``
  542. ## is read) then line will be set to ``""``.
  543. ## The partial line **will be lost**.
  544. ##
  545. ## The ``maxLength`` parameter determines the maximum amount of characters
  546. ## that can be read. The result is truncated after that.
  547. ##
  548. ## **Warning**: The ``Peek`` flag is not yet implemented.
  549. ##
  550. ## **Warning**: ``recvLine`` on unbuffered sockets assumes that the protocol
  551. ## uses ``\r\L`` to delimit a new line.
  552. assert SocketFlag.Peek notin flags ## TODO:
  553. # TODO: Optimise this
  554. var resString = newFutureVar[string]("asyncnet.recvLine")
  555. resString.mget() = ""
  556. await socket.recvLineInto(resString, flags, maxLength)
  557. result = resString.mget()
  558. proc listen*(socket: AsyncSocket, backlog = SOMAXCONN) {.tags: [
  559. ReadIOEffect].} =
  560. ## Marks ``socket`` as accepting connections.
  561. ## ``Backlog`` specifies the maximum length of the
  562. ## queue of pending connections.
  563. ##
  564. ## Raises an OSError error upon failure.
  565. if listen(socket.fd, backlog) < 0'i32: raiseOSError(osLastError())
  566. proc bindAddr*(socket: AsyncSocket, port = Port(0), address = "") {.
  567. tags: [ReadIOEffect].} =
  568. ## Binds ``address``:``port`` to the socket.
  569. ##
  570. ## If ``address`` is "" then ADDR_ANY will be bound.
  571. var realaddr = address
  572. if realaddr == "":
  573. case socket.domain
  574. of AF_INET6: realaddr = "::"
  575. of AF_INET: realaddr = "0.0.0.0"
  576. else:
  577. raise newException(ValueError,
  578. "Unknown socket address family and no address specified to bindAddr")
  579. var aiList = getAddrInfo(realaddr, port, socket.domain)
  580. if bindAddr(socket.fd, aiList.ai_addr, aiList.ai_addrlen.SockLen) < 0'i32:
  581. freeaddrinfo(aiList)
  582. raiseOSError(osLastError())
  583. freeaddrinfo(aiList)
  584. when defined(posix):
  585. proc connectUnix*(socket: AsyncSocket, path: string): owned(Future[void]) =
  586. ## Binds Unix socket to `path`.
  587. ## This only works on Unix-style systems: Mac OS X, BSD and Linux
  588. when not defined(nimdoc):
  589. let retFuture = newFuture[void]("connectUnix")
  590. result = retFuture
  591. proc cb(fd: AsyncFD): bool =
  592. let ret = SocketHandle(fd).getSockOptInt(cint(SOL_SOCKET), cint(SO_ERROR))
  593. if ret == 0:
  594. retFuture.complete()
  595. return true
  596. elif ret == EINTR:
  597. return false
  598. else:
  599. retFuture.fail(newException(OSError, osErrorMsg(OSErrorCode(ret))))
  600. return true
  601. var socketAddr = makeUnixAddr(path)
  602. let ret = socket.fd.connect(cast[ptr SockAddr](addr socketAddr),
  603. (sizeof(socketAddr.sun_family) + path.len).SockLen)
  604. if ret == 0:
  605. # Request to connect completed immediately.
  606. retFuture.complete()
  607. else:
  608. let lastError = osLastError()
  609. if lastError.int32 == EINTR or lastError.int32 == EINPROGRESS:
  610. addWrite(AsyncFD(socket.fd), cb)
  611. else:
  612. retFuture.fail(newException(OSError, osErrorMsg(lastError)))
  613. proc bindUnix*(socket: AsyncSocket, path: string) {.
  614. tags: [ReadIOEffect].} =
  615. ## Binds Unix socket to `path`.
  616. ## This only works on Unix-style systems: Mac OS X, BSD and Linux
  617. when not defined(nimdoc):
  618. var socketAddr = makeUnixAddr(path)
  619. if socket.fd.bindAddr(cast[ptr SockAddr](addr socketAddr),
  620. (sizeof(socketAddr.sun_family) + path.len).SockLen) != 0'i32:
  621. raiseOSError(osLastError())
  622. elif defined(nimdoc):
  623. proc connectUnix*(socket: AsyncSocket, path: string): owned(Future[void]) =
  624. ## Binds Unix socket to `path`.
  625. ## This only works on Unix-style systems: Mac OS X, BSD and Linux
  626. discard
  627. proc bindUnix*(socket: AsyncSocket, path: string) =
  628. ## Binds Unix socket to `path`.
  629. ## This only works on Unix-style systems: Mac OS X, BSD and Linux
  630. discard
  631. proc close*(socket: AsyncSocket) =
  632. ## Closes the socket.
  633. if socket.closed: return
  634. defer:
  635. socket.fd.AsyncFD.closeSocket()
  636. socket.closed = true # TODO: Add extra debugging checks for this.
  637. when defineSsl:
  638. if socket.isSsl:
  639. let res = SSL_shutdown(socket.sslHandle)
  640. SSL_free(socket.sslHandle)
  641. if res == 0:
  642. discard
  643. elif res != 1:
  644. raiseSSLError()
  645. when defineSsl:
  646. proc wrapSocket*(ctx: SslContext, socket: AsyncSocket) =
  647. ## Wraps a socket in an SSL context. This function effectively turns
  648. ## ``socket`` into an SSL socket.
  649. ##
  650. ## **Disclaimer**: This code is not well tested, may be very unsafe and
  651. ## prone to security vulnerabilities.
  652. socket.isSsl = true
  653. socket.sslContext = ctx
  654. socket.sslHandle = SSL_new(socket.sslContext.context)
  655. if socket.sslHandle == nil:
  656. raiseSSLError()
  657. socket.bioIn = bioNew(bioSMem())
  658. socket.bioOut = bioNew(bioSMem())
  659. sslSetBio(socket.sslHandle, socket.bioIn, socket.bioOut)
  660. proc wrapConnectedSocket*(ctx: SslContext, socket: AsyncSocket,
  661. handshake: SslHandshakeType,
  662. hostname: string = "") =
  663. ## Wraps a connected socket in an SSL context. This function effectively
  664. ## turns ``socket`` into an SSL socket.
  665. ## ``hostname`` should be specified so that the client knows which hostname
  666. ## the server certificate should be validated against.
  667. ##
  668. ## This should be called on a connected socket, and will perform
  669. ## an SSL handshake immediately.
  670. ##
  671. ## **Disclaimer**: This code is not well tested, may be very unsafe and
  672. ## prone to security vulnerabilities.
  673. wrapSocket(ctx, socket)
  674. case handshake
  675. of handshakeAsClient:
  676. if hostname.len > 0 and not isIpAddress(hostname):
  677. # Set the SNI address for this connection. This call can fail if
  678. # we're not using TLSv1+.
  679. discard SSL_set_tlsext_host_name(socket.sslHandle, hostname)
  680. sslSetConnectState(socket.sslHandle)
  681. of handshakeAsServer:
  682. sslSetAcceptState(socket.sslHandle)
  683. proc getSockOpt*(socket: AsyncSocket, opt: SOBool, level = SOL_SOCKET): bool {.
  684. tags: [ReadIOEffect].} =
  685. ## Retrieves option ``opt`` as a boolean value.
  686. var res = getSockOptInt(socket.fd, cint(level), toCInt(opt))
  687. result = res != 0
  688. proc setSockOpt*(socket: AsyncSocket, opt: SOBool, value: bool,
  689. level = SOL_SOCKET) {.tags: [WriteIOEffect].} =
  690. ## Sets option ``opt`` to a boolean value specified by ``value``.
  691. var valuei = cint(if value: 1 else: 0)
  692. setSockOptInt(socket.fd, cint(level), toCInt(opt), valuei)
  693. proc isSsl*(socket: AsyncSocket): bool =
  694. ## Determines whether ``socket`` is a SSL socket.
  695. socket.isSsl
  696. proc getFd*(socket: AsyncSocket): SocketHandle =
  697. ## Returns the socket's file descriptor.
  698. return socket.fd
  699. proc isClosed*(socket: AsyncSocket): bool =
  700. ## Determines whether the socket has been closed.
  701. return socket.closed
  702. when not defined(testing) and isMainModule:
  703. type
  704. TestCases = enum
  705. HighClient, LowClient, LowServer
  706. const test = HighClient
  707. when test == HighClient:
  708. proc main() {.async.} =
  709. var sock = newAsyncSocket()
  710. await sock.connect("irc.freenode.net", Port(6667))
  711. while true:
  712. let line = await sock.recvLine()
  713. if line == "":
  714. echo("Disconnected")
  715. break
  716. else:
  717. echo("Got line: ", line)
  718. asyncCheck main()
  719. elif test == LowClient:
  720. var sock = newAsyncSocket()
  721. var f = connect(sock, "irc.freenode.net", Port(6667))
  722. f.callback =
  723. proc (future: Future[void]) =
  724. echo("Connected in future!")
  725. for i in 0 .. 50:
  726. var recvF = recv(sock, 10)
  727. recvF.callback =
  728. proc (future: Future[string]) =
  729. echo("Read ", future.read.len, ": ", future.read.repr)
  730. elif test == LowServer:
  731. var sock = newAsyncSocket()
  732. sock.bindAddr(Port(6667))
  733. sock.listen()
  734. proc onAccept(future: Future[AsyncSocket]) =
  735. let client = future.read
  736. echo "Accepted ", client.fd.cint
  737. var t = send(client, "test\c\L")
  738. t.callback =
  739. proc (future: Future[void]) =
  740. echo("Send")
  741. client.close()
  742. var f = accept(sock)
  743. f.callback = onAccept
  744. var f = accept(sock)
  745. f.callback = onAccept
  746. runForever()