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: nexmark.rs
    path of file: ./repos/arroyo/arroyo-connectors/src
    
    the code of the file until where you have to start completion: use anyhow::{anyhow, bail};
use arroyo_rpc::api_types::connections::FieldType::Primitive;
use arroyo_rpc::api_types::connections::{
    ConnectionProfile, ConnectionSchema, ConnectionType, FieldType, SourceFieldType, StructType,
    TestSourceMessage,
};
use arroyo_rpc::OperatorConfig;
use axum::response::sse::Event;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::convert::Infallible;
use std::str::FromStr;
use typify::import_types;

use crate::{nullable_field, pull_opt, source_field, Connection, Connector, EmptyConfig};

const TABLE_SCHEMA: &str = include_str!("../../connector-schemas/nexmark/table.json");
const ICON: &str = include_str!("../resources/nexmark.svg");

import_types!(schema = "../connector-schemas/nexmark/table.json");

pub fn nexmark_schema() -> ConnectionSchema {
    use arroyo_rpc::api_types::connections::PrimitiveType::*;
    ConnectionSchema {
        format: None,
        bad_data: None,
        framing: None,
        struct_name: Some("arroyo_types::nexmark::Event".to_string()),
        fields: vec![
            nullable_field(
                "person",
                SourceFieldType {
                    r#type: FieldType::Struct(StructType {
                        name: Some("arroyo_types::nexmark::Person".to_string()),
                        fields: vec![
                            source_field("id", Primitive(Int64)),
                            source_field("name", Primitive(String)),
                            source_field("email_address", Primitive(String)),
                            source_field("credit_card", Primitive(String)),
                            source_field("city", Primitive(String)),
                            source_field("state", Primitive(String)),
                            source_field("datetime", Primitive(UnixMillis)),
                            source_field("extra", Primitive(String)),
                        ],
                    }),
                    sql_name: None,
                },
            ),
            nullable_field(
                "bid",
                SourceFieldType {
                    r#type: FieldType::Struct(S
    
pub fn nexmark_schema() -> ConnectionSchema {
    use arroyo_rpc::api_types::connections::PrimitiveType::*;
    ConnectionSchema {
        format: None,
        bad_data: None,
        framing: None,
        struct_name: Some("arroyo_types::nexmark::Event".to_string()),
        fields: vec![
            nullable_field(
                "person",
                SourceFieldType {
                    r#type: FieldType::Struct(StructType {
                        name: Some("arroyo_types::nexmark::Person".to_string()),
                        fields: vec![
                            source_field("id", Primitive(Int64)),
                            source_field("name", Primitive(String)),
                            source_field("email_address", Primitive(String)),
                            source_field("credit_card", Primitive(String)),
                            source_field("city", Primitive(String)),
                            source_field("state", Primitive(String)),
                            source_field("datetime", Primitive(UnixMillis)),
                            source_field("extra", Primitive(String)),
                        ],
                    }),
                    sql_name: None,
                },
            ),
            nullable_field(
                "bid",
                SourceFieldType {
                    r#type: FieldType::Struct(StructType {
                        name: Some("arroyo_types::nexmark::Bid".to_string()),
                        fields: vec![
                            source_field("auction", Primitive(Int64)),
                            source_field("bidder", Primitive(Int64)),
                            source_field("price", Primitive(Int64)),
                            source_field("channel", Primitive(String)),
                            source_field("url", Primitive(String)),
                            source_field("datetime", Primitive(UnixMillis)),
                            source_field("extra", Primitive(String)),
                        ],
                    }),
                    sql_name: None,
                },
            ),
            nullable_field(
                "auction",
                SourceFieldType {
                    r#type: FieldType::Struct(StructType {
                        name: Some("arroyo_types::nexmark::Auction".to_string()),
                        fields: vec![
                            source_field("id", Primitive(Int64)),
                            source_field("description", Primitive(String)),
                            source_field("item_name", Primitive(String)),
                            source_field("initial_bid", Primitive(Int64)),
                            source_field("reserve", Primitive(Int64)),
                            source_field("datetime", Primitive(UnixMillis)),
                            source_field("expires", Primitive(UnixMillis)),
                            source_field("seller", Primitive(Int64)),
                            source_field("category", Primitive(Int64)),
                            source_field("extra", Primitive(String)),
                        ],
                    }),
                    sql_name: None,
                },
            ),
        ],
        definition: None,
        inferred: None,
    }
}