Squashed commit of the following:

commit 67d2559ad253bd3723db73b156d394307741131c
Author: Elliott Williams <[email protected]>
Date:   Mon Oct 9 12:11:56 2017 -0500

    Config updates

commit 46c94fc0b02bb16945a7dbfa0ef0a9babc656e8a
Author: Elliott Williams <[email protected]>
Date:   Mon Oct 9 12:10:38 2017 -0500

    Adoption of new connection objects

    In general, this entails not passing around a `connection` object but rather a
    `ConfigProtocol` signal producer, which can be used to get connection objectss
    ad hoc. Under the new connection design, there isn't a `sharedConnection`
    singleton object floating around.

commit e91f69690ea9c8442676ed3bb2a50cbe2ecaee7b
Author: Elliott Williams <[email protected]>
Date:   Mon Oct 9 12:09:29 2017 -0500

    Rewrite of Connection to be a flatMap of the current configuration

commit db48963010591242dcd6c147050f2b746a15b556
Author: Elliott Williams <[email protected]>
Date:   Mon Oct 9 12:08:25 2017 -0500

    Added agency switcher, presented from POIViewController

Proper.xcodeproj/project.pbxproj

@@ -60,7 +60,6 @@
  26C7FB1C1ED16D6F006B7101 /* POIViewController.swift in Sources */ = {isa = PBXBuildFile; fileRef = 26E7B5441DAFC0AD00AE50E1 /* POIViewController.swift */; };
  26C7FB1D1ED16D6F006B7101 /* POITableViewController.swift in Sources */ = {isa = PBXBuildFile; fileRef = 263D36361DB2784700FC7FF5 /* POITableViewController.swift */; };
  26C7FB1E1ED16D6F006B7101 /* StationViewController.swift in Sources */ = {isa = PBXBuildFile; fileRef = 26A424B51BD2C1320035754F /* StationViewController.swift */; };
- 26C7FB1F1ED16D6F006B7101 /* StartListViewController.swift in Sources */ = {isa = PBXBuildFile; fileRef = 264782551C2C6F67009D9AD0 /* StartListViewController.swift */; };
  26C7FB201ED16D6F006B7101 /* ArrivalsTableViewController.swift in Sources */ = {isa = PBXBuildFile; fileRef = 26A8D6011D484944000434F9 /* ArrivalsTableViewController.swift */; };
  26C7FB211ED16D6F006B7101 /* RouteViewController.swift in Sources */ = {isa = PBXBuildFile; fileRef = 26AE681B1D601D0B0031E534 /* RouteViewController.swift */; };
  26C7FB221ED16D6F006B7101 /* StationTableViewController.swift in Sources */ = {isa = PBXBuildFile; fileRef = 26AE681D1D612B810031E534 /* StationTableViewController.swift */; };
@@ -893,7 +892,6 @@
  26C7FB531ED16D6F006B7101 /* TransitNavigationItem.swift in Sources */,
  26C7FB491ED16D6F006B7101 /* StationUpcomingTableViewCell.swift in Sources */,
  26C7FB341ED16D6F006B7101 /* MutableVehicle.swift in Sources */,
- 26C7FB1F1ED16D6F006B7101 /* StartListViewController.swift in Sources */,
  26C7FB501ED16D6F006B7101 /* StationTableViewHeaderFooterView.swift in Sources */,
  26C7FB481ED16D6F006B7101 /* POIStationAnnotation.swift in Sources */,
  26C7FB211ED16D6F006B7101 /* RouteViewController.swift in Sources */,

Proper/Configuration/BartConfig.swift

@@ -38,7 +38,7 @@ struct BartAgencyConfig: AgencyConfig {
 }
 
 struct BartConnectionConfig: ConnectionConfig {
- let server = URL(string: "ws:https://irene.local:32772/ws")!
+ let server = URL(string: "ws:https://transit.emw.ms/api/bart")!
  let realm = "realm1"
  let scheduleService = "providence"
 }

Proper/Configuration/ConfigProtocols.swift

@@ -16,6 +16,7 @@ protocol ConfigProtocol {
  var connection: ConnectionConfig { get }
  var logging: LoggingConfig { get }
  var ui: UIConfig { get }
+ var id: String { get }
 
  static var id: String { get }
  static func make() -> Self
@@ -41,6 +42,8 @@ protocol ConnectionConfig {
  var server: URL { get }
  var realm: String { get }
  var scheduleService: String { get }
+ var hashed: AnyHashable { get }
+ func makeConnection() -> ConnectionSP
 }
 
 protocol LoggingConfig {
@@ -53,5 +56,13 @@ protocol UIConfig {
 }
 
 extension ConfigProtocol {
+ var id: String { return Self.id }
  static func make() -> Self { return Self() }
 }
+
+extension ConnectionConfig {
+ var hashed: AnyHashable { return server.hashValue ^ realm.hashValue ^ scheduleService.hashValue }
+ func makeConnection() -> ConnectionSP {
+ return Connection.makeFromConfig(connectionConfig: self)
+ }
+}

Proper/Configuration/ConfigSwitcher.swift

@@ -11,13 +11,14 @@ import ReactiveSwift
 import Result
 
 typealias ConfigProperty = MutableProperty<ConfigProtocol>
+typealias ConfigSP = SignalProducer<ConfigProtocol, NoError>
 
 struct Config {
  typealias DefaultConfig = CitybusConfig
 
- static let knownConfigurations: [ConfigProtocol.Type] = [
- CitybusConfig.self,
- BartConfig.self
+ static let knownConfigurations: [ConfigProtocol] = [
+ CitybusConfig(),
+ BartConfig()
  ]
 
  static let shared: ConfigProperty = {
@@ -41,10 +42,13 @@ private extension Config {
  guard let id = UserDefaults.standard.string(forKey: "selectedConfig") else {
  return nil
  }
- return knownConfigurations.first(where: { $0.id == id })?.make()
+ let config = knownConfigurations.first(where: { $0.id == id })
+ NSLog("[Config.stored] using \(config ?? DefaultConfig())")
+ return config
  }
 
  static func store(config: ConfigProtocol) {
  UserDefaults.standard.set(type(of: config).id, forKey: "selectedConfig")
+ NSLog("[Config.store] set to \(config)")
  }
 }

Proper/Configuration/SharedConfig.swift

@@ -19,7 +19,8 @@ struct SharedLoggingConfig: LoggingConfig {
  "MDWamp.callWithSignal",
  "Connection.connectionProducer",
  "Connection.subscribe",
- "MutableRoute.producer"
+ "MutableRoute.producer",
+ "Timetable",
  ])
  let logJSON = false
 }

Proper/Connection/CachedConnection.swift

@@ -10,40 +10,49 @@ import Foundation
 import ReactiveSwift
 import Runes
 
-class CachedConnection<C: ConnectionType>: ConnectionType {
- let connection: C
+protocol CachedConnectionProtocol: class {
+ var lastEventCache: LastEventCache { get }
+}
+
+extension CachedConnectionProtocol {
+ func updatingCache(withEventsFrom producer: SignalProducer<TopicEvent, ProperError>, for topic: String) ->
+ SignalProducer<TopicEvent, ProperError>
+ {
+ return producer.on(terminated: { [weak self] in self?.lastEventCache.expire(topic: topic) },
+ value: { [weak self] in _ = self?.lastEventCache.store(event: $0, from: topic) })
+ }
+
+ func cacheLookup(_ proc: String, _ args: WampArgs, _ kwargs: WampKwargs) -> EventProducer {
+ return EventProducer { observer, _ in
+ if proc == "meta.last_event", let topic = args[safe: 0] as? String, let originator = args[safe: 1] as? String {
+ self.lastEventCache.lastEvent(from: originator, sentIn: topic).apply(observer.send)
+ }
+ observer.sendCompleted()
+ }.logEvents(identifier: "\(String(describing: Self.self)).cacheLookup(proc: \(proc))",
+ events: Set([.value]), logger: logSignalEvent)
+ }
+}
+
+@available(deprecated, message: "Connection implementations should conform to CachedConnectionProtocol")
+class CachedConnection<C: ConnectionType>: ConnectionType, CachedConnectionProtocol {
+ let inner: C
  let lastEventCache = LastEventCache()
 
  init(_ connection: C) {
- self.connection = connection
+ self.inner = connection
  }
 
  // Returns a producer which will check the cache before calling the underlying connection.
  func call(_ proc: String, with args: WampArgs, kwargs: WampKwargs) -> EventProducer {
  let hit = cacheLookup(proc, args, kwargs)
- let miss = connection.call(proc, with: args, kwargs: kwargs)
+ let miss = inner.call(proc, with: args, kwargs: kwargs)
  return SignalProducer<EventProducer, ProperError>([hit, miss])
  .flatten(.concat).take(first: 1)
  }
 
  // Returns a producer of topic events that will update the cache, and will trigger a cache void when the connection
  // terminates.
  func subscribe(to topic: String) -> EventProducer {
- return connection.subscribe(to: topic)
- .on(terminated: { [weak self] in self?.lastEventCache.expire(topic: topic) },
- value: { [weak self] in _ = self?.lastEventCache.store(event: $0, from: topic) })
- }
-
- private func cacheLookup(_ proc: String, _ args: WampArgs, _ kwargs: WampKwargs) -> EventProducer {
- return EventProducer { observer, _ in
- if proc == "meta.last_event", let topic = args[safe: 0] as? String, let originator = args[safe: 1] as? String {
- self.lastEventCache.lastEvent(from: originator, sentIn: topic).apply(observer.send)
- }
- observer.sendCompleted()
- }
+ return updatingCache(withEventsFrom: inner.subscribe(to: topic), for: topic)
  }
 }
-
-extension Connection {
- static let cachedInstance = CachedConnection(Connection.sharedInstance)
-}

Proper/Connection/Connection.swift

@@ -11,84 +11,110 @@ import MDWamp
 import ReactiveSwift
 import Result
 
-class Connection: NSObject {
- typealias ConnectionProducer = SignalProducer<MDWamp, ProperError>
-
- static var sharedInstance = Connection.init()
-
- // MARK: Private
- fileprivate var config = Config.shared
- fileprivate lazy var connection = MutableProperty<MDWamp?>(nil)
- fileprivate let disposable = ScopedDisposable(CompositeDisposable())
-
- override init() {
+typealias ConnectionSP = SignalProducer<Connection, ProperError>
+
+class Connection: NSObject, CachedConnectionProtocol {
+ fileprivate let wamp: MDWamp
+ fileprivate let (signal, observer) = Signal<Connection, ProperError>.pipe()
+ let lastEventCache = LastEventCache()
+ fileprivate static var connections: [AnyHashable: Connection] = [:]
+
+ var isConnected: Bool { return wamp.isConnected() }
+ fileprivate var isConnecting: Bool = false
+ fileprivate var connectionTimeoutAction: ScopedDisposable<AnyDisposable>? = nil
+
+ fileprivate init(connectionConfig config: ConnectionConfig) {
+ let transport = MDWampTransportWebSocket(server: config.server,
+ protocolVersions: [kMDWampProtocolWamp2msgpack, kMDWampProtocolWamp2json])
+ NSLog("[Connection.init] Created transport for \(config.server)")
+ wamp = MDWamp(transport: transport, realm: config.realm, delegate: nil)
  super.init()
+ wamp.delegate = self
+ }
 
- // Connect the wamp instance based on the latest configuration, and disconnect from old instances.
- disposable.inner += config.producer
- .map(makeConnection)
- .map(Optional.init)
- .combinePrevious(nil)
- .startWithValues { prev, next in
- prev?.disconnect()
- next?.connect()
+ func connect() -> Disposable {
+ if !isConnecting {
+ isConnecting = true
+ wamp.connect()
+ NSLog("[Connection.connect]")
  }
+ if let timeout = QueueScheduler.main.schedule(after: Date(timeIntervalSinceNow: 10), action: { [weak self] in
+ self?.observer.send(error: .timeout(rpc: "unable to connect"))
+ }) {
+ connectionTimeoutAction = ScopedDisposable(timeout)
+ }
+// return ActionDisposable { [weak self] in self?.disconnect() }
+ return ActionDisposable { }
  }
-}
-
-// MARK: - Connection forming
-extension Connection {
 
- // MARK: Private
-
- /// Returns an `MDWamp` object created using `config`. Disconnects the previous connection, if it exists, and connects
- /// the returned connection.
- fileprivate func makeConnection(config: ConfigProtocol) -> MDWamp {
- let ws = MDWampTransportWebSocket(server: config.connection.server,
- protocolVersions: [kMDWampProtocolWamp2msgpack, kMDWampProtocolWamp2json])
- return MDWamp(transport: ws, realm: config.connection.realm, delegate: self)!
+ func disconnect() {
+ NSLog("[Connection.disconnect]")
+ connectionTimeoutAction?.dispose()
+ wamp.disconnect()
  }
 }
 
-// MARK: - Communication methods
 extension Connection: ConnectionType {
  /// Subscribe to `topic` and forward parsed events. Disposing of signals created from this method will unsubscribe
  /// `topic`.
  func subscribe(to topic: String) -> EventProducer {
- return connection.producer.skipNil()
- .map { wamp in wamp.subscribeWithSignal(topic) }
- .flatten(.latest)
+ let subscription = wamp
+ .subscribeWithSignal(topic)
  .map { TopicEvent.parse(from: topic, event: $0) }
  .unwrapOrSendFailure(ProperError.eventParseFailure)
  .logEvents(identifier: "Connection.subscribe", logger: logSignalEvent)
+ return updatingCache(withEventsFrom: subscription, for: topic)
  }
 
  /// Call `proc` and forward the result. Disposing the signal created will cancel the RPC call.
  func call(_ proc: String, with args: WampArgs = [], kwargs: WampKwargs = [:]) -> EventProducer {
- return connection.producer.skipNil()
- .map({ $0.callWithSignal(proc, args, kwargs, [:])
- .timeout(after: 10.0, raising: .timeout(rpc: proc), on: QueueScheduler.main) })
- .flatten(.latest)
+ let cache = cacheLookup(proc, args, kwargs)
+ let server = wamp.callWithSignal(proc, args, kwargs, [:])
+ .timeout(after: 10.0, raising: .timeout(rpc: proc), on: QueueScheduler.main)
  .map { TopicEvent.parse(fromRPC: proc, args, kwargs, $0) }
  .unwrapOrSendFailure(ProperError.eventParseFailure)
+ .logEvents(identifier: "Connection.call(proc: \(proc))", events: Set([.starting, .value, .failed]),
+ logger: logSignalEvent)
+ return SignalProducer<EventProducer, ProperError>([cache, server])
+ .flatten(.concat).take(first: 1)
  }
 }
 
 // MARK: - MDWampClientDelegate
 extension Connection: MDWampClientDelegate {
- func mdwamp(_ wamp: MDWamp!, sessionEstablished info: [AnyHashable: Any]!) {
- NSLog("[Connection] Session established")
- connection.swap(wamp)
+ func mdwamp(_ wamp: MDWamp!, sessionEstablished info: [AnyHashable : Any]!) {
+ isConnecting = false
+ connectionTimeoutAction?.dispose()
+ observer.send(value: self)
  }
 
- func mdwamp(_ wamp: MDWamp!, closedSession code: Int, reason: String!, details: WampKwargs!) {
- NSLog("[Connection] Session closed, code=\(code) reason=\(reason)")
-
- if code == MDWampConnectionCloseCode.closed.rawValue {
- // We're switching connections.
+ func mdwamp(_ wamp: MDWamp!, closedSession code: Int, reason: String!, details: [AnyHashable : Any]!) {
+ connectionTimeoutAction?.dispose()
+ if MDWampConnectionCloseCode(rawValue: code) == .closed {
+ observer.sendCompleted()
  } else {
- // TODO show connection error
- fatalError(reason)
+ observer.send(error: .connectionLost(reason: reason))
+ }
+ }
+}
+
+// MARK: - Factory
+extension Connection {
+ static func makeFromConfig(connectionConfig config: ConnectionConfig) -> SignalProducer<Connection, ProperError> {
+ return SignalProducer { observer, disposable in
+ let saved = connections[config.hashed]
+ switch saved {
+ case .some(let saved) where saved.isConnected:
+ observer.send(value: saved)
+ case .some(let saved): // otherwise
+ disposable += saved.signal.observe(observer)
+ disposable += saved.connect()
+ case .none:
+ let connection = Connection(connectionConfig: config)
+ disposable += connection.signal.observe(observer)
+ disposable += connection.connect()
+ connections[config.hashed] = connection
+ }
  }
  }
 }
@@ -100,7 +126,6 @@ extension MDWamp {
  -> SignalProducer<MDWampResult, ProperError>
  {
  return SignalProducer<MDWampResult, ProperError> { observer, _ in
- NSLog("[Connection] Calling \(procUri)")
  self.call(procUri, args: args, kwArgs: argsKw, options: options) { result, error in
  if error != nil {
  observer.send(error: .mdwampError(topic: procUri, object: error))

Proper/Controllers/AgencyTableViewController.swift

@@ -7,3 +7,60 @@
 //
 
 import Foundation
+import UIKit
+import ReactiveSwift
+
+class AgencyTableViewController: UITableViewController {
+ let configurations: [ConfigProtocol]
+ let configProperty: MutableProperty<ConfigProtocol>
+
+ init(configurations: [ConfigProtocol], configProperty: MutableProperty<ConfigProtocol>) {
+ self.configurations = configurations
+ self.configProperty = configProperty
+ super.init(style: .grouped)
+
+ tableView.register(DisposableCell.self, forCellReuseIdentifier: "cell")
+ navigationItem.title = "Transit Agencies"
+ navigationItem.rightBarButtonItem = UIBarButtonItem(barButtonSystemItem: .done, target: self, action: #selector(done(_:)))
+ }
+
+ @IBAction func done(_ sender: AnyObject) {
+ dismiss(animated: true, completion: nil)
+ }
+
+ required init?(coder aDecoder: NSCoder) {
+ fatalError("init(coder:) has not been implemented")
+ }
+}
+
+private class DisposableCell: UITableViewCell {
+ var disposable = ScopedDisposable(CompositeDisposable())
+ override func prepareForReuse() {
+ disposable = ScopedDisposable(CompositeDisposable())
+ }
+}
+
+extension AgencyTableViewController {
+ override func numberOfSections(in tableView: UITableView) -> Int {
+ return 1
+ }
+
+ override func tableView(_ tableView: UITableView, numberOfRowsInSection section: Int) -> Int {
+ return configurations.count
+ }
+
+ override func tableView(_ tableView: UITableView, didSelectRowAt indexPath: IndexPath) {
+ configProperty.swap(configurations[indexPath.row])
+ tableView.deselectRow(at: indexPath, animated: true)
+ }
+
+ override func tableView(_ tableView: UITableView, cellForRowAt indexPath: IndexPath) -> UITableViewCell {
+ let cell = tableView.dequeueReusableCell(withIdentifier: "cell", for: indexPath) as! DisposableCell
+ let config = configurations[indexPath.row]
+ cell.textLabel?.text = config.agency.name
+
+ cell.disposable += configProperty.map({ $0.id == config.id ? UITableViewCellAccessoryType.checkmark : .none })
+ .producer.startWithValues { cell.accessoryType = $0 }
+ return cell
+ }
+}