void ConnectMQTT() {
  static const char *password = MQTT_PASSWORD;
  static const char *user = MQTT_USER;
  while (!mqtt_client.connect("brewhouse", user, password)) {
    Serial.print(".");
    delay(1000);
  }

  char topic[30];
  strcpy(topic,TOPIC_ROOT);
  strcat(topic,BOIL_SETPOINT_TOPIC);
  mqtt_client.subscribe(topic);
}

void MessageReceived(char* topic, byte* payload, unsigned int length) {
  char buf[30];
  strcpy(buf,TOPIC_ROOT);
  strcat(buf,BOIL_SETPOINT_TOPIC);

  char msg[length+1];

  for (int i=0;i<length;i++) {
    msg[i] = (char)payload[i];
  }
  msg[length] = 0;
  
  if (strcmp(topic, buf) == 0) {
    cJSON *monitor_json = cJSON_Parse(msg);
    const cJSON *name = NULL;
    const cJSON *setting = NULL;
    const cJSON *unit = NULL;

    // Update PWM setpoint.
    setting = cJSON_GetObjectItemCaseSensitive(monitor_json, "setpoint");
    KettleDuty = setting->valueint;
  }
}

static void SendSensorData() {

  char *string = NULL;
  cJSON *entity = NULL;
  cJSON *setpoint = NULL;
  cJSON *units = NULL;

  cJSON *monitor = cJSON_CreateObject();
  cJSON_AddStringToObject(monitor, "entity", "boil_kettle");
  cJSON_AddNumberToObject(monitor, "setpoint", KettleDuty);
  cJSON_AddStringToObject(monitor, "units", "%");
  char *msg = cJSON_Print(monitor);

  char topic[30];
  strcpy(topic,TOPIC_ROOT);
  strcat(topic,BOIL_ACTUAL_TOPIC);

  mqtt_client.publish(topic, msg);

}