package com.codespair.kafka.navigator.kafkanavigatorbe.resources;

import com.codespair.kafka.navigator.kafkanavigatorbe.kafka.BusClientMetaData;
import com.codespair.kafka.navigator.kafkanavigatorbe.model.KMetric;
import com.codespair.kafka.navigator.kafkanavigatorbe.kafka.jmx.KafkaJMX;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.Optional;
import java.util.Set;

@Slf4j
@RestController
@RequestMapping("/api/bus")
public class BusController {

  private KafkaJMX kafkaJMX;
  private BusClientMetaData busClientMetaData;

  public BusController(KafkaJMX kafkaJMX, BusClientMetaData busClientMetaData) {
    this.kafkaJMX = kafkaJMX;
    this.busClientMetaData = busClientMetaData;
  }

  @CrossOrigin(origins="*")
  @PostMapping("/connect")
  public ResponseEntity connect(@RequestBody String jmxServerUrl) {
    log.info("Received request to connect, trying to connect with kafka jmx: {}", jmxServerUrl);
    Optional<List<String>> domainList =  kafkaJMX.connect(jmxServerUrl);
    if(domainList.isPresent())
    {
      return ResponseEntity.ok().body(domainList.get());
    }
    return ResponseEntity.noContent().build();
  }


  @CrossOrigin(origins="*")
  @PostMapping("/k-client-metrics")
  public ResponseEntity kClientMetrics(@RequestBody List<String> hosts) {
    log.info("Received request to connect, trying to connect and get metrics list of kafka client: {}", hosts);
    Set<KMetric> metricSet = busClientMetaData.initializeClient(hosts);
    return ResponseEntity.ok().body(metricSet);
  }
}

@@ -1047,36 +1047,41 @@
 nal%3C
-List%3CString%3E%3E domainList =  
+Boolean%3E connected = Optional.of(
 kafk
@@ -1110,101 +1110,126 @@
 Url)
+)
 ;%0A    
-if(domainList.isPresent())
+Optional%3CInteger%3E brokerId = Optional.empty();
 %0A    
-%7B%0A      return ResponseEntity.ok().body(domainList.get()
+if(connected.get()) %7B%0A      brokerId = kafkaJMX.getBrokerId(
 );%0A 
@@ -1263,34 +1263,45 @@
 ity.
-noContent().build(
+ok().body(brokerId.orElse(-1)
 );%0A  %7D%0A%0A
 %0A  @
@@ -1296,17 +1296,16 @@
 );%0A  %7D%0A%0A
-%0A
   @Cross

connect endpoint first release.