Merge pull request #1031 from qvacua/pipe

Use pipe for IPC

NvimView/Sources/NvimView/Defs.swift

@@ -5,7 +5,7 @@
 
 import Foundation
 
-struct Defs {
+enum Defs {
   static let loggerSubsystem = "com.qvacua.NvimView"
 
   enum LoggerCategory {

NvimView/Sources/NvimView/KeyUtils.swift

@@ -5,7 +5,7 @@
 
 import Cocoa
 
-final class KeyUtils {
+enum KeyUtils {
   static func isControlCode(key: String) -> Bool {
     guard key.count == 1 else {
       return false

NvimView/Sources/NvimView/NvimView+RemoteOptions.swift

@@ -58,10 +58,18 @@ extension NvimView {
 
     switch option {
     case let .guifont(fontSpec):
-      command = self.api.setOptionValue(name: "guifont", value: .string(fontSpec), opts: ["scope": .string("global")])
+      command = self.api.setOptionValue(
+        name: "guifont",
+        value: .string(fontSpec),
+        opts: ["scope": .string("global")]
+      )
 
     case let .guifontWide(fontSpec):
-      command = self.api.setOptionValue(name: "guifontwide", value: .string(fontSpec), opts: ["scope": .string("global")])
+      command = self.api.setOptionValue(
+        name: "guifontwide",
+        value: .string(fontSpec),
+        opts: ["scope": .string("global")]
+      )
     }
 
     command

NvimView/Sources/NvimView/NvimView+Resize.swift

@@ -71,102 +71,82 @@ extension NvimView {
 
     self.log.info("NVIM_LISTEN_ADDRESS=\(sockPath)")
 
+    let inPipe: Pipe, outPipe: Pipe, errorPipe: Pipe
     do {
-      try self.bridge.runLocalServerAndNvim(width: size.width, height: size.height)
+      (inPipe, outPipe, errorPipe) = try self.bridge.runLocalServerAndNvim(
-    }
+        width: size.width, height: size.height
-    catch let err as RxNeovimApi.Error {
+      )
+    } catch let err as RxNeovimApi.Error {
       self.eventsSubject.onNext(.ipcBecameInvalid(
         "Could not launch neovim (\(err))."
       ))
-    }
+
-    catch {
+      return
+    } catch {
       self.eventsSubject.onNext(.ipcBecameInvalid(
         "Could not launch neovim."
       ))
-    }
 
-    // Wait for listen and socket creation to occur
-    let timeout = Duration.seconds(4)
-    let start_time = ContinuousClock.now
-    while !FileManager.default.fileExists(atPath: sockPath) {
-      usleep(1000)
-      if ContinuousClock.now - start_time > timeout {
-        self.eventsSubject.onNext(.ipcBecameInvalid(
-          "Timeout waiting for neovim."
-        ))
       return
     }
-    }
 
     // We wait here, since the user of NvimView cannot subscribe
     // on the Completable. We could demand that the user call launchNeoVim()
     // by themselves, but...
-    try?
+
-      self.api.run(at: sockPath)
+    // See https://neovim.io/doc/user/ui.html#ui-startup for startup sequence
+    // When we call nvim_command("autocmd VimEnter * call rpcrequest(1, 'vimenter')")
+    // Neovim will send us a vimenter request and enter a blocking state.
+    // We do some autocmd setup and send a response to exit the blocking state in
+    // NvimView.swift
+    try? self.api.run(inPipe: inPipe, outPipe: outPipe, errorPipe: errorPipe)
       .andThen(
-        self.api.getApiInfo()
+        self.api.getApiInfo(errWhenBlocked: false)
-          .do(onError: { err in
+          .flatMapCompletable { value in
-            throw RxNeovimApi.Error
-              .exception(message: "Could not connect to neovim (\(err)).")
-          })
-          .map {
-          value in
             guard let info = value.arrayValue,
                   info.count == 2,
-                let channel = info[0].int32Value
+                  let channel = info[0].int32Value,
+                  let dict = info[1].dictionaryValue,
+                  let version = dict["version"]?.dictionaryValue,
+                  let major = version["major"]?.intValue,
+                  let minor = version["minor"]?.intValue
             else {
-            throw RxNeovimApi.Error
+              throw RxNeovimApi.Error.exception(message: "Could not convert values to api info.")
-              .exception(message: "Could not convert values to api info.")
             }
 
-          return channel
+            guard (major >= kMinAlphaVersion && minor >= kMinMinorVersion) || major >=
-        }.flatMapCompletable {
+              kMinMajorVersion
-          // FIXME: make lua
-          self.api.exec2(src: """
-          ":augroup vimr
-          ":augroup!
-          :autocmd VimEnter * call rpcnotify(\($0), 'autocommand', 'vimenter')
-          :autocmd BufWinEnter * call rpcnotify(\(
-            $0
-          ), 'autocommand', 'bufwinenter', str2nr(expand('<abuf>')))
-          :autocmd BufWinEnter * call rpcnotify(\(
-            $0
-          ), 'autocommand', 'bufwinleave', str2nr(expand('<abuf>')))
-          :autocmd TabEnter * call rpcnotify(\(
-            $0
-          ), 'autocommand', 'tabenter', str2nr(expand('<abuf>')))
-          :autocmd BufWritePost * call rpcnotify(\(
-            $0
-          ), 'autocommand', 'bufwritepost', str2nr(expand('<abuf>')))
-          :autocmd BufEnter * call rpcnotify(\(
-            $0
-          ), 'autocommand', 'bufenter', str2nr(expand('<abuf>')))
-          :autocmd DirChanged * call rpcnotify(\(
-            $0
-          ), 'autocommand', 'dirchanged', expand('<afile>'))
-          :autocmd ColorScheme * call rpcnotify(\($0), 'autocommand', 'colorscheme', \
-              get(nvim_get_hl(0, {'id': hlID('Normal')}), 'fg', -1), \
-              get(nvim_get_hl(0, {'id': hlID('Normal')}), 'bg', -1), \
-              get(nvim_get_hl(0, {'id': hlID('Visual')}), 'fg', -1), \
-              get(nvim_get_hl(0, {'id': hlID('Visual')}), 'bg', -1), \
-              get(nvim_get_hl(0, {'id': hlID('Directory')}), 'fg', -1))
-          :autocmd ExitPre * call rpcnotify(\($0), 'autocommand', 'exitpre')
-          :autocmd BufModifiedSet * call rpcnotify(\($0), 'autocommand', 'bufmodifiedset', \
-              str2nr(expand('<abuf>')), getbufinfo(str2nr(expand('<abuf>')))[0].changed)
-          :let g:gui_vimr = 1
-          ":augroup END
-          """, opts: [:]).asCompletable()
-
-            .andThen(
-              self.sourceFileUrls.reduce(Completable.empty()) { prev, url in
-                prev
-                  .andThen(
-                    self.api.exec2(src: "source \(url.shellEscapedPath)", opts: ["output": true])
-                      .map {
-                        retval in
-                        guard let output_value = retval["output"] ?? retval["output"],
-                              let output = output_value.stringValue
             else {
+              self.eventsSubject.onNext(.ipcBecameInvalid(
+                "Incompatible neovim version \(major).\(minor)"
+              ))
+              throw RxNeovimApi.Error.exception(message: "Incompatible neovim version.")
+            }
+
+            return self.api.exec2(src: """
+            let g:gui_vimr = 1
+            autocmd ExitPre * call rpcnotify(\(channel), 'autocommand', 'exitpre')
+            autocmd VimEnter * call rpcnotify(\(channel), 'autocommand', 'vimenter')
+            autocmd ColorScheme * call rpcnotify(\(channel), 'autocommand', 'colorscheme', get(nvim_get_hl(0, {'id': hlID('Normal')}), 'fg', -1), get(nvim_get_hl(0, {'id': hlID('Normal')}), 'bg', -1), get(nvim_get_hl(0, {'id': hlID('Visual')}), 'fg', -1), get(nvim_get_hl(0, {'id': hlID('Visual')}), 'bg', -1), get(nvim_get_hl(0, {'id': hlID('Directory')}), 'fg', -1))
+            autocmd VimEnter * call rpcrequest(\(channel), 'vimenter')
+            """, opts: [:], errWhenBlocked: false)
+              .asCompletable()
+          }
+      )
+      .andThen(
+        self.api.uiAttach(width: size.width, height: size.height, options: [
+          "ext_linegrid": true,
+          "ext_multigrid": false,
+          "ext_tabline": MessagePackValue(self.usesCustomTabBar),
+          "rgb": true,
+        ])
+      )
+      .andThen(
+        self.sourceFileUrls.reduce(.empty()) { prev, url in
+          prev.andThen(
+            self.api.exec2(src: "source \(url.shellEscapedPath)", opts: ["output": true])
+              .map { retval in
+                guard let output = retval["output"]?.stringValue else {
                   throw RxNeovimApi.Error
                     .exception(message: "Could not convert values to output.")
                 }
@@ -176,15 +156,8 @@ extension NvimView {
           )
         }
       )
-            .andThen(self.api.uiAttach(width: size.width, height: size.height, options: [
-              "ext_linegrid": true,
-              "ext_multigrid": false,
-              "ext_tabline": MessagePackValue(self.usesCustomTabBar),
-              "rgb": true,
-            ]))
       .andThen(self.api.subscribe(event: NvimView.rpcEventName))
-        }
+      .wait()
-      ).wait()
   }
 
   private func randomEmoji() -> String {

NvimView/Sources/NvimView/NvimView.swift

@@ -182,8 +182,19 @@ public final class NvimView: NSView, NSUserInterfaceValidations, NSTextInputClie
     self.api.msgpackRawStream
       .subscribe(onNext: { [weak self] msg in
         switch msg {
+        case let .request(msgid, method, _):
+          // See https://neovim.io/doc/user/ui.html#ui-startup
+          // "vimenter" RPC request will be sent to us
+          // which is the result of
+          // nvim_command("autocmd VimEnter * call rpcrequest(1, 'vimenter')") in
+          // NvimView+Resize.swift
+          // This is the only request sent from Neovim to the UI, afaics.
+          guard method == "vimenter" else { break }
+          self?.log.debug("Processing blocking vimenter request")
+          self?.setupAutocmdsAndSendResponse(forMsgid: msgid)
+
         case let .notification(method, params):
-          self?.log.debug("NOTIFICATION: \(method): \(params)")
+          self?.log.trace("NOTIFICATION: \(method): \(params)")
 
           if method == NvimView.rpcEventName {
             self?.eventsSubject.onNext(.rpcEvent(params))
@@ -346,4 +357,33 @@ public final class NvimView: NSView, NSUserInterfaceValidations, NSTextInputClie
 
   private var _linespacing = NvimView.defaultLinespacing
   private var _characterspacing = NvimView.defaultCharacterspacing
+
+  private func setupAutocmdsAndSendResponse(forMsgid msgid: UInt32) {
+    self.api.getApiInfo(errWhenBlocked: false)
+      .flatMapCompletable { value in
+        guard let info = value.arrayValue,
+              info.count == 2,
+              let channel = info[0].int32Value
+        else {
+          throw RxNeovimApi.Error.exception(message: "Could not convert values to api info.")
+        }
+
+        // swiftformat:disable all
+        return self.api.exec2(src: """
+        autocmd BufWinEnter * call rpcnotify(\(channel), 'autocommand', 'bufwinenter', str2nr(expand('<abuf>')))
+        autocmd BufWinLeave * call rpcnotify(\(channel), 'autocommand', 'bufwinleave', str2nr(expand('<abuf>')))
+        autocmd TabEnter * call rpcnotify(\(channel), 'autocommand', 'tabenter', str2nr(expand('<abuf>')))
+        autocmd BufWritePost * call rpcnotify(\(channel), 'autocommand', 'bufwritepost', str2nr(expand('<abuf>')))
+        autocmd BufEnter * call rpcnotify(\(channel), 'autocommand', 'bufenter', str2nr(expand('<abuf>')))
+        autocmd DirChanged * call rpcnotify(\( channel), 'autocommand', 'dirchanged', expand('<afile>'))
+        autocmd BufModifiedSet * call rpcnotify(\(channel), 'autocommand', 'bufmodifiedset', str2nr(expand('<abuf>')), getbufinfo(str2nr(expand('<abuf>')))[0].changed)
+        """, opts: [:], errWhenBlocked: false)
+        // swiftformat:enable all
+          .asCompletable()
+          .andThen(
+            self.api.sendResponse(msgid: msgid, error: .nil, result: .nil)
+          )
+      }
+      .subscribe().disposed(by: self.disposeBag)
+  }
 }

NvimView/Sources/NvimView/UiBridge.swift

@@ -7,13 +7,12 @@ import Commons
 import Foundation
 import MessagePack
 import os
-import RxPack
-import RxSwift
 import RxNeovim
 import RxPack
+import RxSwift
 
 let kMinAlphaVersion = 0
-let kMinMinorVersion = 10
+let kMinMinorVersion = 9
 let kMinMajorVersion = 1
 
 final class UiBridge {
@@ -39,11 +38,11 @@ final class UiBridge {
     }
   }
 
-  func runLocalServerAndNvim(width: Int, height: Int) throws {
+  func runLocalServerAndNvim(width: Int, height: Int) throws -> (Pipe, Pipe, Pipe) {
     self.initialWidth = width
     self.initialHeight = height
 
-    try self.launchNvimUsingLoginShellEnv()
+    return try self.launchNvimUsingLoginShellEnv()
   }
 
   func quit() -> Completable {
@@ -70,7 +69,7 @@ final class UiBridge {
     self.nvimServerProc?.terminate()
   }
 
-  private func launchNvimUsingLoginShellEnv() throws {
+  private func launchNvimUsingLoginShellEnv() throws -> (Pipe, Pipe, Pipe) {
     var env = self.envDict
     env["NVIM_LISTEN_ADDRESS"] = self.listenAddress
 
@@ -85,9 +84,7 @@ final class UiBridge {
     process.standardOutput = outPipe
     process.currentDirectoryPath = self.cwd.path
 
-    if self.nvimBinary != "",
+    if self.nvimBinary != "", FileManager.default.fileExists(atPath: self.nvimBinary) {
-       FileManager.default.fileExists(atPath: self.nvimBinary)
-    {
       process.launchPath = self.nvimBinary
     } else {
       // We know that NvimServer is there.
@@ -97,11 +94,7 @@ final class UiBridge {
     }
     process.environment = env
 
-    process
+    process .arguments = ["--embed"] + self.nvimArgs
-      .arguments =
-      ["--embed",
-       "--listen",
-       self.listenAddress] + self.nvimArgs
 
     self.log.debug(
       "Launching NvimServer \(String(describing: process.launchPath)) with args: \(String(describing: process.arguments))"
@@ -113,71 +106,8 @@ final class UiBridge {
         .exception(message: "Could not run neovim process.")
     }
 
-    try self.doInitialVersionCheck(inPipe: inPipe, outPipe: outPipe)
-
     self.nvimServerProc = process
-  }
+    return (inPipe, outPipe, errorPipe)
-
-  private func doInitialVersionCheck(inPipe: Pipe, outPipe: Pipe) throws {
-
-    // Construct Msgpack query for api info
-    let packed = pack(
-      [
-        .uint(RxMsgpackRpc.MessageType.request.rawValue),
-        .uint(UInt64(0)),
-        .string("nvim_get_api_info"),
-        .array([]),
-      ]
-    )
-
-    try inPipe.fileHandleForWriting.write(contentsOf: packed)
-
-    // Read responses from the pipe back
-    var accumulatedData : Data = Data()
-    var values : [MessagePackValue] = []
-    var remainderData: Data? = nil
-    while (true) {
-      let data = outPipe.fileHandleForReading.availableData
-      if data.count == 0 {
-        break
-      }
-      accumulatedData.append(data)
-      
-      try (values, remainderData) = RxMsgpackRpc.unpackAllWithReminder(accumulatedData)
-      
-      if let remainderData { accumulatedData = remainderData }
-      else { accumulatedData.count = 0 }
-      
-      if values.count > 0 {
-        break
-      }
-    }
-
-    // Validate version response
-    guard values.count >= 1,
-          let firstResponse = values[0].arrayValue,
-          firstResponse.count == 4,
-          let rawType = firstResponse[0].uint64Value,
-          let type = RxMsgpackRpc.MessageType(rawValue: rawType),
-          type == RxMsgpackRpc.MessageType.response /* this is a response */,
-          let msgId = firstResponse[1].uint64Value,
-          msgId == 0 /* no confusion on stream */,
-          firstResponse[2] == nil /* no error */,
-          let info = firstResponse[3].arrayValue /* response value */,
-          info.count == 2,
-          let dict = info[1].dictionaryValue,
-          let version = dict["version"]?.dictionaryValue,
-          let major = version["major"]?.intValue,
-          let minor = version["minor"]?.intValue
-    else {
-      throw RxNeovimApi.Error
-        .exception(message: "Could not convert values to api info.")
-    }
-    guard (major >= kMinAlphaVersion && minor >= kMinMinorVersion) || major >= kMinMajorVersion
-    else {
-      throw RxNeovimApi.Error
-        .exception(message: "Incompatible neovim version.")
-    }
   }
 
   private func interactive(for shell: URL) -> Bool {

RxPack/Package.swift

@@ -4,6 +4,7 @@ import PackageDescription
 
 let package = Package(
   name: "RxPack",
+  platforms: [.macOS(.v13)],
   products: [
     .library(name: "RxPack", targets: ["RxPack"]),
     .library(name: "RxNeovim", targets: ["RxNeovim"]),
@@ -11,14 +12,12 @@ let package = Package(
   dependencies: [
     .package(url: "https://github.com/ReactiveX/RxSwift", from: "6.6.0"),
     .package(url: "https://github.com/a2/MessagePack.swift", .upToNextMinor(from: "4.0.0")),
-    .package(url: "https://github.com/IBM-Swift/BlueSocket", from: "2.0.2"),
     .package(url: "https://github.com/Quick/Nimble", from: "13.0.0"),
   ],
   targets: [
     .target(name: "RxPack", dependencies: [
       .product(name: "RxSwift", package: "RxSwift"),
       .product(name: "MessagePack", package: "MessagePack.swift"),
-      .product(name: "Socket", package: "BlueSocket"),
     ]),
     .testTarget(name: "RxPackTests", dependencies: [
       "RxPack",

RxPack/Sources/RxNeovim/RxNeovimApi.swift

@@ -34,7 +34,9 @@ public final class RxNeovimApi {
 
   public var msgpackRawStream: Observable<RxMsgpackRpc.Message> { self.msgpackRpc.stream }
 
-  public func run(at path: String) -> Completable { self.msgpackRpc.run(at: path) }
+  public func run(inPipe: Pipe, outPipe: Pipe, errorPipe: Pipe) -> Completable {
+    self.msgpackRpc.run(inPipe: inPipe, outPipe: outPipe, errorPipe: errorPipe)
+  }
 
   public func stop() -> Completable { self.msgpackRpc.stop() }
 
@@ -64,6 +66,10 @@ public final class RxNeovimApi {
       }
   }
 
+  public func sendResponse(msgid: UInt32, error: Value, result: Value) -> Completable {
+    self.msgpackRpc.response(msgid: msgid, error: error, result: result)
+  }
+
   public init() {}
 
   private let msgpackRpc = RxMsgpackRpc(queueQos: .userInteractive)

RxPack/Sources/RxPack/RxMsgpackRpc.swift

@@ -4,7 +4,6 @@
 import Foundation
 import MessagePack
 import RxSwift
-import Socket
 
 public final class RxMsgpackRpc {
   public static let defaultReadBufferSize = 10240
@@ -21,6 +20,7 @@ public final class RxMsgpackRpc {
     case response(msgid: UInt32, error: Value, result: Value)
     case notification(method: String, params: [Value])
     case error(value: Value, msg: String)
+    case request(msgid: UInt32, method: String, params: [Value])
   }
 
   public struct Response {
@@ -67,23 +67,14 @@ public final class RxMsgpackRpc {
     )
   }
 
-  public func run(
+  public func run(inPipe: Pipe, outPipe: Pipe, errorPipe: Pipe) -> Completable {
-    at path: String,
+    self.inPipe = inPipe
-    readBufferSize: Int = RxMsgpackRpc.defaultReadBufferSize
+    self.outPipe = outPipe
-  ) -> Completable {
+    self.errorPipe = errorPipe
-    Completable.create { completable in
-      self.queue.async {
-        do {
-          try self.socket = Socket.create(family: .unix, type: .stream, proto: .unix)
-          self.socket?.readBufferSize = readBufferSize
-          try self.socket?.connect(to: path)
-          self.setUpThreadAndStartReading()
-        } catch {
-          self.socket = nil
-          self.streamSubject.onError(Error(msg: "Could not get socket", cause: error))
-          completable(.error(Error(msg: "Could not get socket at \(path)", cause: error)))
-        }
 
+    return Completable.create { completable in
+      self.queue.async {
+        self.startReading()
         completable(.completed)
       }
 
@@ -94,7 +85,7 @@ public final class RxMsgpackRpc {
   public func stop() -> Completable {
     Completable.create { completable in
       self.queue.async {
-        self.cleanUpAndCloseSocket()
+        self.cleanUp()
         completable(.completed)
       }
 
@@ -102,6 +93,38 @@ public final class RxMsgpackRpc {
     }
   }
 
+  public func response(msgid: UInt32, error: Value, result: Value) -> Completable {
+    Completable.create { completable in
+      self.queue.async {
+        let packed = pack(
+          [
+            .uint(MessageType.response.rawValue),
+            .uint(UInt64(msgid)),
+            error,
+            result,
+          ]
+        )
+
+        do {
+          try self.inPipe?.fileHandleForWriting.write(contentsOf: packed)
+          completable(.completed)
+        } catch {
+          self.streamSubject.onError(Error(
+            msg: "Could not write to socket for msg id: \(msgid)", cause: error
+          ))
+
+          completable(.error(Error(
+            msg: "Could not write to socket for msg id: \(msgid)", cause: error
+          )))
+
+          return
+        }
+      }
+
+      return Disposables.create()
+    }
+  }
+
   public func request(
     method: String,
     params: [Value],
@@ -121,33 +144,10 @@ public final class RxMsgpackRpc {
           ]
         )
 
-        if self.socket?.remoteConnectionClosed == true {
-          single(.failure(Error(
-            msg: "Connection stopped, but trying to send a request with msg id \(msgid)"
-          )))
-          return
-        }
-
-        guard let socket = self.socket else {
-          single(.failure(Error(
-            msg: "Socket is invalid, but trying to send a request with " +
-              "msg id \(msgid): \(method) with \(params)"
-          )))
-          return
-        }
-
         if expectsReturnValue { self.singles[msgid] = single }
 
         do {
-          var remainder: Data? = packed
+          try self.inPipe?.fileHandleForWriting.write(contentsOf: packed)
-          while let dataToSend = remainder {
-            let writtenBytes = try socket.write(from: dataToSend)
-            if writtenBytes < dataToSend.count {
-              remainder = packed.suffix(from: writtenBytes)
-            } else {
-              remainder = nil
-            }
-          }
         } catch {
           self.streamSubject.onError(Error(
             msg: "Could not write to socket for msg id: \(msgid)", cause: error
@@ -169,10 +169,13 @@ public final class RxMsgpackRpc {
 
   private var nextMsgid: UInt32 = 0
 
-  private var socket: Socket?
   private let queue: DispatchQueue
   private let dataQueue: DispatchQueue
 
+  private var inPipe: Pipe?
+  private var outPipe: Pipe?
+  private var errorPipe: Pipe?
+
   private var singles: [UInt32: SingleResponseObserver] = [:]
 
   private let streamSubject = PublishSubject<Message>()
@@ -181,52 +184,40 @@ public final class RxMsgpackRpc {
     Response(msgid: msgid, error: .nil, result: .nil)
   }
 
-  private func cleanUpAndCloseSocket() {
+  private func cleanUp() {
+    self.inPipe = nil
+    self.outPipe = nil
+    self.errorPipe = nil
+
     self.streamSubject.onCompleted()
 
     self.singles.forEach { _, single in single(.failure(Error(msg: "Socket closed"))) }
-    self.singles.removeAll()
-
-    self.socket?.close()
   }
 
-  private func setUpThreadAndStartReading() {
+  private func startReading() {
     self.dataQueue.async { [unowned self] in
-      guard let socket = self.socket else { return }
+      var readData: Data
-
       var dataToUnmarshall = Data(capacity: Self.defaultReadBufferSize)
       repeat {
         do {
-          var readData = Data(capacity: Self.defaultReadBufferSize)
+          guard let buffer = self.outPipe?.fileHandleForReading.availableData else { break }
-          let readBytes = try socket.read(into: &readData)
+          readData = buffer
 
-          if readBytes > 0 {
+          if readData.count > 0 {
             dataToUnmarshall.append(readData)
-            let (values, remainderData) = try RxMsgpackRpc.unpackAllWithReminder(dataToUnmarshall)
+            let (values, remainderData) = try self.unpackAllWithReminder(dataToUnmarshall)
             if let remainderData { dataToUnmarshall = remainderData }
             else { dataToUnmarshall.count = 0 }
 
             values.forEach(self.processMessage)
-          } else if readBytes == 0 {
-            if socket.remoteConnectionClosed {
-              self.queue.async { self.cleanUpAndCloseSocket() }
-              return
           }
-
-            continue
-          }
-        } catch let error as Socket.Error {
-          self.streamSubject.onError(Error(msg: "Could not read from socket", cause: error))
-          self.queue.async { self.cleanUpAndCloseSocket() }
-          return
         } catch {
-          self.streamSubject.onNext(
+          self.streamSubject.onError(Error(msg: "Could not read from pipe", cause: error))
-            .error(value: .nil, msg: "Data from socket could not be unpacked")
-          )
-          self.queue.async { self.cleanUpAndCloseSocket() }
-          return
         }
-      } while self.socket?.remoteConnectionClosed == false
+      } while readData.count > 0
+
+      self.streamSubject.onNext(.notification(method: "autocommand", params: ["exitpre"]))
+      self.cleanUp()
     }
   }
 
@@ -286,15 +277,20 @@ public final class RxMsgpackRpc {
       self.streamSubject.onNext(.notification(method: method, params: params))
 
     case .request:
-      self.streamSubject.onNext(.error(
+      guard let msgid = array[1].uint32Value, let method = array[2].stringValue,
-        value: unpacked,
+            let params = array[3].arrayValue
-        msg: "Got message type request from remote"
+      else {
-      ))
+        return
+      }
+
+      self.streamSubject.onNext(.request(msgid: msgid, method: method, params: params))
       return
     }
   }
 
-  public static func unpackAllWithReminder(_ data: Data) throws -> (values: [Value], remainder: Data?) {
+  private func unpackAllWithReminder(_ data: Data) throws
+    -> (values: [Value], remainder: Data?)
+  {
     var values = [Value]()
     var remainderData: Data?
 

VimR.xcworkspace/xcshareddata/swiftpm/Package.resolved

@@ -1,14 +1,5 @@
 {
   "pins" : [
-    {
-      "identity" : "bluesocket",
-      "kind" : "remoteSourceControl",
-      "location" : "https://github.com/IBM-Swift/BlueSocket",
-      "state" : {
-        "revision" : "7b23a867008e0027bfd6f4d398d44720707bc8ca",
-        "version" : "2.0.4"
-      }
-    },
     {
       "identity" : "cwlcatchexception",
       "kind" : "remoteSourceControl",

VimR/VimR/MainWindow+Actions.swift

@@ -13,7 +13,6 @@ import Workspace
 
 extension MainWindow {
   func rpcEventAction(params rawParams: [MessagePackValue]) {
-    Swift.print("################### \(rawParams)")
     guard rawParams.count > 0 else { return }
 
     guard let strEvent = rawParams[0].stringValue,

bin/build_nvimserver.sh

@@ -21,11 +21,13 @@ main() {
 
   if [[ "${for_dev}" == true ]]; then
     pushd ./Neovim >/dev/null
-      make CMAKE_BUILD_TYPE=Release
+      mkdir -p ./build/install
+      make CMAKE_BUILD_TYPE=Release CMAKE_EXTRA_FLAGS="-DCMAKE_INSTALL_PREFIX=./install"
+      make install
     popd >/dev/null
 
-    cp ./Neovim/build/bin/nvim "${resources_folder}/NvimServer"
+    cp ./Neovim/build/install/bin/nvim "${resources_folder}/NvimServer"
-    cp -r ./Neovim/build/runtime "${resources_folder}"
+    cp -r ./Neovim/build/install/share/nvim/runtime "${resources_folder}"
   else
     ./bin/neovim/bin/build_neovim.sh
     pushd ./Neovim/build >/dev/null