You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: sse.rs
    path of file: ./repos/arroyo/arroyo-connectors/src
    
    the code of the file until where you have to start completion: use std::collections::HashMap;
use std::convert::Infallible;
use std::time::Duration;

use anyhow::{anyhow, bail};
use arroyo_rpc::{var_str::VarStr, OperatorConfig};
use arroyo_types::string_to_map;
use axum::response::sse::Event;
use eventsource_client::Client;
use futures::StreamExt;
use tokio::sync::mpsc::Sender;
use typify::import_types;

use arroyo_rpc::api_types::connections::{
    ConnectionProfile, ConnectionSchema, ConnectionType, TestSourceMessage,
};
use serde::{Deserialize, Serialize};

use crate::{pull_opt, Connection, EmptyConfig};

use super::Connector;

const TABLE_SCHEMA: &str = include_str!("../../connector-schemas/sse/table.json");

import_types!(schema = "../connector-schemas/sse/table.json", convert = { {type = "string", format = "var-str"} = VarStr });
const ICON: &str = include_str!("../resources/sse.svg");

pub struct SSEConnector {}

impl Connector for SSEConnector {
    type ProfileT = EmptyConfig;

    type TableT = SseTable;

    fn name(&self) -> &'static str {
        "sse"
    }

    fn metadata(&self) -> arroyo_rpc::api_types::connections::Connector {
        arroyo_rpc::api_types::connections::Connector {
            id: "sse".to_string(),
            name: "Server-Sent Events".to_string(),
            icon: ICON.to_string(),
            description: "Connect to a SSE/EventSource server".to_string(),
            enabled: true,
            source: true,
            sink: false,
            testing: true,
            hidden: false,
            custom_schemas: true,
            con
    
    async fn test_internal(&self) -> anyhow::Result<()> {
        let mut client = eventsource_client::ClientBuilder::for_url(&self.config.endpoint)
            .map_err(|_| anyhow!("Endpoint URL is invalid"))?;

        let headers = string_to_map(
            &self
                .config
                .headers
                .as_ref()
                .map(|s| s.sub_env_vars())
                .transpose()?
                .unwrap_or("".to_string()),
            ':',
        )
        .ok_or_else(|| anyhow!("Headers are invalid; should be comma-separated pairs"))?;

        for (k, v) in headers {
            client = client
                .header(&k, &v)
                .map_err(|_| anyhow!("Invalid header '{}: {}'", k, v))?;
        }

        let mut stream = client.build().stream();

        let timeout = Duration::from_secs(30);

        let message = TestSourceMessage {
            error: false,
            done: false,
            message: "Constructed SSE client".to_string(),
        };
        self.tx
            .send(Ok(Event::default().json_data(message).unwrap()))
            .await
            .unwrap();

        tokio::select! {
            val = stream.next() => {
                // TODO: validate schema
                match val {
                    Some(Ok(_)) => {
                        let message = TestSourceMessage {
                            error: false,
                            done: false,
                            message: "Received message from SSE server".to_string()
                        };
                        self.tx.send(Ok(Event::default().json_data(message).unwrap())).await.unwrap();
                    }
                    Some(Err(e)) => {
                        bail!("Received error from server: {:?}", e);
                    }
                    None => {
                        bail!("Server closed connection");
                    }
                }
            }
            _ = tokio::time::sleep(timeout) => {
                bail!("Did not receive any messages after 30 seconds");
            }
        };

        Ok(())
    }