vis: handle datasource websocket errors more gracefully (#1598)

packages/ott-vis-datasource/src/datasource.ts

@@ -16,6 +16,7 @@ import { Observable, lastValueFrom, merge } from "rxjs";
 
 export class DataSource extends DataSourceApi<MyQuery, MyDataSourceOptions> {
 	baseUrl: string;
+	socket: WebSocket | null = null;
 
 	constructor(instanceSettings: DataSourceInstanceSettings<MyDataSourceOptions>) {
 		super(instanceSettings);
@@ -38,9 +39,12 @@ export class DataSource extends DataSourceApi<MyQuery, MyDataSourceOptions> {
 					frame.addField({ name: "direction", type: FieldType.string });
 
 					const base = this.baseUrl.replace(/^http/, "ws");
-					const ws = new WebSocket(`${base}/state/stream`);
 
-					ws.addEventListener("message", msg => {
+					const open = (e: Event) => {
+						console.log("WebSocket opened", e);
+					};
+
+					const message = (msg: MessageEvent) => {
 						const event = JSON.parse(msg.data);
 						frame.add(event);
 
@@ -49,16 +53,57 @@ export class DataSource extends DataSourceApi<MyQuery, MyDataSourceOptions> {
 							key: frame.refId,
 							state: LoadingState.Streaming,
 						});
-					});
+					};
 
-					ws.addEventListener("error", err => {
+					const error = (err: Event) => {
 						console.error("WebSocket error", err);
 						subscriber.error(err);
-					});
+					};
 
-					ws.addEventListener("close", () => {
-						subscriber.complete();
-					});
+					const close = (e: CloseEvent) => {
+						// subscriber.complete();
+						console.warn("WebSocket closed", e);
+						if (e.code === 4132) {
+							console.info("We aborted the connection.");
+							return;
+						}
+						console.log("Datasource reconnecting...");
+						setTimeout(() => {
+							this.socket = new WebSocket(`${base}/state/stream`);
+							addListeners(this.socket);
+						}, 1000);
+					};
+
+					function addListeners(ws: WebSocket) {
+						ws.addEventListener("open", open);
+						ws.addEventListener("message", message);
+						ws.addEventListener("error", error);
+						ws.addEventListener("close", close);
+					}
+					function removeListeners(ws: WebSocket) {
+						ws.removeEventListener("open", open);
+						ws.removeEventListener("message", message);
+						ws.removeEventListener("error", error);
+						ws.removeEventListener("close", close);
+					}
+
+					function teardown(this: DataSource) {
+						if (!this.socket) {
+							return;
+						}
+						this.socket.close(4132);
+						removeListeners(this.socket);
+						window.removeEventListener("beforeunload", teardown);
+					}
+					subscriber.add(teardown.bind(this));
+					window.addEventListener("beforeunload", teardown);
+
+					if (this.socket) {
+						teardown.call(this);
+					}
+
+					this.socket = new WebSocket(`${base}/state/stream`);
+					addListeners(this.socket);
 				});
 			}
 

packages/ott-vis-panel/src/components/CorePanel.tsx

@@ -32,13 +32,30 @@ const getStyles = () => {
 	};
 };
 
-export const CorePanel: React.FC<Props> = ({ options, data, width, height }) => {
+export const CorePanel: React.FC<Props> = props => {
+	const { data } = props;
+
+	if (data.state === LoadingState.Error) {
+		return <CoreError data={data} />;
+	}
+
+	return <CoreData {...props} />;
+};
+
+/**
+ * Shown when the data source is in a nominal state.
+ */
+const CoreData: React.FC<Props> = ({ options, data, width, height }) => {
 	const styles = useStyles2(getStyles);
 
 	const stateSeries = data.series[0];
 	const eventBusSeries = data.series[1];
 	const eventBus = useEventBus();
 
+	if (!stateSeries) {
+		console.log("No state series, data:", data);
+	}
+
 	const systemState: SystemState = useMemo(() => {
 		return options.useSampleData
 			? sampleSystemState
@@ -116,6 +133,20 @@ export const CorePanel: React.FC<Props> = ({ options, data, width, height }) =>
 	);
 };
 
+/**
+ * Shown when the data source is in an error state.
+ */
+const CoreError: React.FC<Pick<Props, "data">> = ({ data }) => {
+	return (
+		<div>
+			Errors:
+			{data.errors?.map((e, i) => (
+				<div key={i}>{e.message}</div>
+			))}
+		</div>
+	);
+};
+
 const Loading: React.FC = () => {
 	return <div style={{ position: "absolute", top: 0, left: 0 }}>Loading...</div>;
 };