       .filter { message in
         return message.topic == topic
       }
-      .debug()
   }
   
 }
    func unsubAckReceived(session: MQTTSession!, msgID: UInt16) {
      print("unsubscribe acknowledged")
    }

  }

  let mqttSession: MQTTSession
  let messages: Observable<Message>

  convenience init(userDefaults: NSUserDefaults = NSUserDefaults.standardUserDefaults()) {
    let host = userDefaults.stringForKey("mqtt_host")!


  func publish(message: Message) -> Promise<Message, HomeClientError> {
    let publish = PromiseSource<Message, HomeClientError>()
    mqttSession.publishData(message.payload, onTopic: message.topic, retain: message.retain, qos: message.mqttQos) { [weak self] error in
      if error != nil {
        publish.reject(HomeClientError(message: error.description))
      } else {
        print("published message '\(message.payloadString)' on topic '\(message.topic)'")
        self?.currentObserver?.on(.Next(message))
        publish.resolve(message)
      }
    }
