void ConnectMQTT() {
  ConfigData config;
  EEPROM.get(ConfAddress, config);
  
  static const char *user = config.mqtt.user;
  static const char *password = config.mqtt.password;
  //config.mqtt.topic.root
  //config.mqtt.broker

  while (!mqtt_client.connect("brewhouse", user, password)) {
    Serial.print(".");
    delay(1000);
  }

  char topic[30];
  strcpy(topic,config.mqtt.topic.root);
  strcat(topic,config.mqtt.topic.boil.setpoint);
  mqtt_client.subscribe(topic);

  strcpy(topic,config.mqtt.topic.root);
  strcat(topic,config.mqtt.topic.mash.setpoint);
  mqtt_client.subscribe(topic);
}

void MessageReceived(char* topic, byte* payload, unsigned int length) {
  ConfigData config;
  EEPROM.get(ConfAddress, config);

  char buf[30];
  strcpy(buf,config.mqtt.topic.root);
  strcat(buf,config.mqtt.topic.boil.setpoint);

  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;
    cJSON_Delete(monitor_json);
  }
}

static void SendSensorData() {
  ConfigData config;
  EEPROM.get(ConfAddress, config);

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

  cJSON *monitor = cJSON_CreateObject();
  cJSON_AddStringToObject(monitor, "entity", config.mqtt.topic.boil.name);
  cJSON_AddNumberToObject(monitor, "setpoint", KettleDuty);
  cJSON_AddStringToObject(monitor, "units", "%");
  char *msg = cJSON_Print(monitor);

  char topic[30];
  strcpy(topic,config.mqtt.topic.root);
  strcat(topic,config.mqtt.topic.boil.sensor);


  mqtt_client.publish(topic, msg);
  cJSON_Delete(monitor);

}