void ConnectMQTT() {
  Serial.println("connecting MQTT...");
  while (!mqtt_client.connect("arduino", "mqtt_user", "4SutKhR2ZEET2IU0PNhH")) {
    Serial.print(".");
    delay(1000);
  }

  Serial.println("\nconnected!");
  mqtt_client.subscribe("brewery/setpoint/bk");
}

void MessageReceived(String &topic, String &payload) {
  Serial.println("incoming: " + topic + " - " + payload);

  /** JSON Parser Setup */
  StaticJsonDocument<200> doc;

  // Deserialize the JSON document
  DeserializationError error = deserializeJson(doc, payload);

  // Test if parsing succeeds.
  if (error) {
    Serial.print(F("deserializeJson() failed: "));
    Serial.println(error.f_str());
    return;
  }
  char buf[30];
  strcpy(buf,TOPIC_PREFIX);
  strcat(buf,BKTOPIC);
  if (topic == buf) {
    // Update PWM setpoint.
    String name = doc["entity"];
    String setting = doc["setpoint"];

    kettle_duty = setting.toInt();
    String unit = doc["units"];

    Serial.println("Updating setpoint for " + name + " to " + setting + " " + unit);
  }
}

void SetupMQTT(String broker) {
  // Note: Local domain names (e.g. "Computer.local" on OSX) are not supported
  // by Arduino. You need to set the IP address directly.
  Serial.println("Setup MQTT client.");
  mqtt_client.begin("192.168.1.198", net);
  mqtt_client.onMessage(MessageReceived);

  ConnectMQTT();
}

static void sendSensorData() {
  Serial.println("Sending data...");

  // NOTE: max message length is 250 bytes.
  StaticJsonDocument<200> doc;

  doc["entity"] = "boil_kettle";
  doc["setpoint"] = kettle_duty;
  doc["units"] = "%";

  String jstr;
  serializeJson(doc, jstr);

  String topic = TOPIC_PREFIX;
  topic += "sensor/boil_kettle";

  mqtt_client.publish(topic, jstr);

}