nim-libp2p/examples/directchat.nim

when not(compileOption("threads")):
  {.fatal: "Please, compile this program with the --threads:on option!".}

import tables, strformat, strutils
import chronos                              # an efficient library for async
import ../libp2p/[switch,                   # manage transports, a single entry point for dialing and listening
                  multistream,              # tag stream with short header to identify it
                  crypto/crypto,            # cryptographic functions
                  errors,                   # error handling utilities
                  protocols/identify,       # identify the peer info of a peer
                  connection,               # create and close stream read / write connections
                  transports/transport,     # listen and dial to other peers using p2p protocol
                  transports/tcptransport,  # listen and dial to other peers using client-server protocol
                  multiaddress,             # encode different addressing schemes. For example, /ip4/7.7.7.7/tcp/6543 means it is using IPv4 protocol and TCP
                  peerinfo,                 # manage the information of a peer, such as peer ID and public / private key
                  peer,                     # Implement how peers interact
                  protocols/protocol,       # define the protocol base type
                  protocols/secure/secure,  # define the protocol of secure connection
                  protocols/secure/secio,   # define the protocol of secure input / output, allows encrypted communication that uses public keys to validate signed messages instead of a certificate authority like in TLS
                  muxers/muxer,             # define an interface for stream multiplexing, allowing peers to offer many protocols over a single connection
                  muxers/mplex/mplex,       # implement stream multiplexing
                  muxers/mplex/types]       # define some contants and message types for stream multiplexing

const ChatCodec = "/nim-libp2p/chat/1.0.0"
const DefaultAddr = "/ip4/127.0.0.1/tcp/55505"

const Help = """
  Commands: /[?|hep|connect|disconnect|exit]
  help: Prints this help
  connect: dials a remote peer
  disconnect: ends current session
  exit: closes the chat
"""

type ChatProto = ref object of LPProtocol
  switch: Switch          # a single entry point for dialing and listening to peer
  transp: StreamTransport # transport streams between read & write file descriptor
  conn: Connection        # create and close read & write stream
  connected: bool         # if the node is connected to another peer
  started: bool           # if the node has started


proc initAddress(T: type MultiAddress, str: string): T =
  let address = MultiAddress.init(str)
  if IPFS.match(address) and matchPartial(multiaddress.TCP, address):
    result = address
  else:
    raise newException(MultiAddressError,
                         "Invalid bootstrap node multi-address")

proc dialPeer(p: ChatProto, address: string) {.async.} =
  let multiAddr = MultiAddress.initAddress(address);
  let parts = address.split("/")
  let remotePeer = PeerInfo.init(parts[^1],
                                 [multiAddr])

  echo &"dialing peer: {multiAddr}"
  p.conn = await p.switch.dial(remotePeer, ChatCodec)
  p.connected = true

proc readAndPrint(p: ChatProto) {.async.} =
  while true:
    while p.connected:
      # TODO: echo &"{p.id} -> "

      echo cast[string](await p.conn.readLp(1024))
    await sleepAsync(100.millis)

proc writeAndPrint(p: ChatProto) {.async.} =
  while true:
    if not p.connected:
      echo "type an address or wait for a connection:"
      echo "type /[help|?] for help"

    let line = await p.transp.readLine()
    if line.startsWith("/help") or line.startsWith("/?") or not p.started:
      echo Help
      continue

    if line.startsWith("/disconnect"):
      echo "Ending current session"
      if p.connected and p.conn.closed.not:
        await p.conn.close()
      p.connected = false
    elif line.startsWith("/connect"):
      if p.connected:
        var yesno = "N"
        echo "a session is already in progress, do you want end it [y/N]?"
        yesno = await p.transp.readLine()
        if yesno.cmpIgnoreCase("y") == 0:
          await p.conn.close()
          p.connected = false
        elif yesno.cmpIgnoreCase("n") == 0:
          continue
        else:
          echo "unrecognized response"
          continue

      echo "enter address of remote peer"
      let address = await p.transp.readLine()
      if address.len > 0:
        await p.dialPeer(address)

    elif line.startsWith("/exit"):
      if p.connected and p.conn.closed.not:
        await p.conn.close()
        p.connected = false

      await p.switch.stop()
      echo "quitting..."
      quit(0)
    else:
      if p.connected:
        await p.conn.writeLp(line)
      else:
        try:
          if line.startsWith("/") and "ipfs" in line:
            await p.dialPeer(line)
        except:
          echo &"unable to dial remote peer {line}"
          echo getCurrentExceptionMsg()

proc readWriteLoop(p: ChatProto) {.async.} =
  asyncCheck p.writeAndPrint() # execute the async function but does not block
  asyncCheck p.readAndPrint()

proc newChatProto(switch: Switch, transp: StreamTransport): ChatProto =
  var chatproto = ChatProto(switch: switch, transp: transp, codec: ChatCodec)

  # create handler for incoming connection
  proc handle(stream: Connection, proto: string) {.async.} =
    if chatproto.connected and not chatproto.conn.closed:
      echo "a chat session is already in progress - disconnecting!"
      await stream.close()
    else:
      chatproto.conn = stream
      chatproto.connected = true

  # assign the new handler
  chatproto.handler = handle
  return chatproto

proc readInput(wfd: AsyncFD) {.thread.} =
  ## This procedure performs reading from `stdin` and sends data over
  ## pipe to main thread.
  let transp = fromPipe(wfd)

  while true:
    let line = stdin.readLine()
    discard waitFor transp.write(line & "\r\n")

proc processInput(rfd: AsyncFD) {.async.} =
  let transp = fromPipe(rfd)

  let seckey = PrivateKey.random(RSA).get()
  let peerInfo = PeerInfo.init(seckey)
  var localAddress = DefaultAddr
  while true:
    echo &"Type an address to bind to or Enter to use the default {DefaultAddr}"
    let a = await transp.readLine()
    try:
      if a.len > 0:
        peerInfo.addrs.add(Multiaddress.init(a))
        break

      peerInfo.addrs.add(Multiaddress.init(localAddress))
      break
    except:
      echo "invalid address"
      localAddress = DefaultAddr
      continue

  # a constructor for building different multiplexers under various connections
  proc createMplex(conn: Connection): Muxer =
    result = newMplex(conn)

  let mplexProvider = newMuxerProvider(createMplex, MplexCodec)
  let transports = @[Transport(TcpTransport.init())]
  let muxers = [(MplexCodec, mplexProvider)].toTable()
  let identify = newIdentify(peerInfo)
  let secureManagers = [(SecioCodec, Secure(newSecio(seckey)))].toTable()
  let switch = newSwitch(peerInfo,
                         transports,
                         identify,
                         muxers,
                         secureManagers)

  let chatProto = newChatProto(switch, transp)
  switch.mount(chatProto)
  let libp2pFuts = await switch.start()
  chatProto.started = true

  let id = peerInfo.peerId.pretty
  echo "PeerID: " & id
  echo "listening on: "
  for a in peerInfo.addrs:
    echo &"{a}/ipfs/{id}"

  await chatProto.readWriteLoop()
  await allFuturesThrowing(libp2pFuts)

proc main() {.async.} =
  let (rfd, wfd) = createAsyncPipe()
  if rfd == asyncInvalidPipe or wfd == asyncInvalidPipe:
    raise newException(ValueError, "Could not initialize pipe!")

  var thread: Thread[AsyncFD]
  thread.createThread(readInput, wfd)

  await processInput(rfd)

when isMainModule: # isMainModule = true when the module is compiled as the main file
  waitFor(main())