95 lines
2.5 KiB
Dart
95 lines
2.5 KiB
Dart
import '../../about/internal/application/ServerProfile.dart';
|
|
import '../parent/ParentViewModel.dart';
|
|
import 'ConnectKafka.dart';
|
|
|
|
class ViewKafka extends ParentViewModel {
|
|
ConnectKafka connection;
|
|
|
|
ViewKafka(super.context, this.connection);
|
|
|
|
/// Looks for the Kafka CLI tools on PATH and in the usual install dirs.
|
|
void detectKafka(ServerProfile server) async {
|
|
if (!await canReach(server, () => detectKafka(server))) return;
|
|
try {
|
|
final status = await getDataManager().kafkaDetect(server);
|
|
connection.onKafkaDetected(status);
|
|
} catch (error) {
|
|
connection.onDetectFailed(error.toString());
|
|
}
|
|
}
|
|
|
|
/// Reads topics, consumer groups and cluster health in one pass. They're
|
|
/// three separate CLI invocations, so this is where they're sequenced.
|
|
void scan(
|
|
ServerProfile server, {
|
|
required String binDir,
|
|
required String suffix,
|
|
required String bootstrap,
|
|
required bool includeInternal,
|
|
}) async {
|
|
try {
|
|
final topics = await getDataManager().kafkaTopics(
|
|
server,
|
|
binDir: binDir,
|
|
suffix: suffix,
|
|
bootstrap: bootstrap,
|
|
includeInternal: includeInternal,
|
|
);
|
|
final groups = await getDataManager().kafkaGroups(
|
|
server,
|
|
binDir: binDir,
|
|
suffix: suffix,
|
|
bootstrap: bootstrap,
|
|
);
|
|
final cluster = await getDataManager().kafkaCluster(
|
|
server,
|
|
binDir: binDir,
|
|
suffix: suffix,
|
|
bootstrap: bootstrap,
|
|
);
|
|
|
|
connection.onScanned(
|
|
KafkaScan(topics: topics, groups: groups, cluster: cluster),
|
|
);
|
|
} catch (error) {
|
|
connection.onScanFailed(error.toString());
|
|
}
|
|
}
|
|
|
|
/// Runs `kafka-broker-api-versions` so the user can see exactly what the
|
|
/// broker says when a scan comes back empty.
|
|
void probeBroker(
|
|
ServerProfile server, {
|
|
required String binDir,
|
|
required String suffix,
|
|
required String bootstrap,
|
|
}) async {
|
|
showLoading("Probing $bootstrap …");
|
|
try {
|
|
final output = await getDataManager().kafkaTest(
|
|
server,
|
|
binDir: binDir,
|
|
suffix: suffix,
|
|
bootstrap: bootstrap,
|
|
);
|
|
closeLoading();
|
|
connection.onProbeOutput(
|
|
'kafka-broker-api-versions · $bootstrap',
|
|
output.isEmpty ? '(no output — broker unreachable?)' : output,
|
|
);
|
|
} catch (error) {
|
|
handleError(
|
|
error,
|
|
() => probeBroker(
|
|
server,
|
|
binDir: binDir,
|
|
suffix: suffix,
|
|
bootstrap: bootstrap,
|
|
),
|
|
() {},
|
|
"Retry",
|
|
);
|
|
}
|
|
}
|
|
}
|