async fn run_server(&mut self, server_behavior: ServerBehavior) -> SocketAddr {
        if self.server_join_handle.is_some() {
            panic!("one `TestFixture` can only run one server at a time");
        }
        self.server_stopper = ServerStopper::new();

        // Set the server to use a channel buffer of half the total events it will emit, unless
        // we're running with no delay between events, in which case set a minimal buffer as we're
        // trying to cause clients to get ejected for lagging.
        let config = Config {
            event_stream_buffer_length: if server_behavior.has_delay_between_events {
                BUFFER_LENGTH
            } else {
                1
            },
            max_concurrent_subscribers: server_behavior
                .max_concurrent_subscribers
                .unwrap_or(Config::default().max_concurrent_subscribers),
            ..Default::default()
        };
        let mut server = EventStreamServer::new(
            config,
            self.storage_dir.path().to_path_buf(),
            self.protocol_version,
            self.deploy_getter.clone(),
        )
        .unwrap();

        self.first_event_id = server.event_indexer.current_index();

        let first_event_id = server.event_indexer.current_index();
        let server_address = server.listening_address;
        let events = self.events.clone();
        let server_stopper = self.server_stopper.clone();

        let join_handle = tokio::spawn(async move {
            let event_count = if server_behavior.repeat_events {
                MAX_EVENT_COUNT
            } else {
                EVENT_COUNT
            };
            for (id, event) in events.iter().cycle().enumerate().take(event_count as usize) {
                if server_stopper.should_stop() {
                    debug!("stopping server early");
                    return;
                }
                server_behavior
                    .wait_for_clients((id as Id).wrapping_add(first_event_id))
                    .await;
                let _ = server.broadcast(event.clone());
                server_behavior.sleep_if_required().await;
            }

            // Keep the server running until told to stop.  Clients connecting from now will only
            // receive keepalives.
            debug!("server finished sending all events");
            server_stopper.wait().await;
            debug!("server stopped");
        });

        self.server_join_handle = Some(join_handle);

        server_address
    }
/// Creates a new `EventStreamServer` and runs it in a tokio task, returning the actual address
/// the server is listening on.
///
/// Only one server can be run at a time; this panics if there is already a server task running.
///
/// The server emits a clone of each of the random events held by the `TestFixture`, in the
/// order in which they're held in the `TestFixture`.
///
/// The server runs until `TestFixture::stop_server()` is called, or the `TestFixture` is
/// dropped.
Creates a new `EventStreamServer` and runs it in a tokio task, returning the actual address
the server is listening on.

Only one server can be run at a time; this panics if there is already a server task running.

The server emits a clone of each of the random events held by the `TestFixture`, in the
order in which they're held in the `TestFixture`.

The server runs until `TestFixture::stop_server()` is called, or the `TestFixture` is
dropped.
Creates a new ` EventStreamServer ` and runs it in a tokio task returning the actual address the server is listening on . Only one server can be run at a time ; this panics if there is already a server task running . The server emits a clone of each of the random events held by the ` TestFixture ` in the order in which they ' re held in the ` TestFixture ` . The server runs until ` TestFixture :: stop_server () ` is called or the ` TestFixture ` is dropped .
async fn run_server(&mut self, server_behavior: ServerBehavior) -> SocketAddr {
        if self.server_join_handle.is_some() {
            panic!("one `TestFixture` can only run one server at a time");
        }
        self.server_stopper = ServerStopper::new();
        // Set the server to use a channel buffer of half the total events it will emit, unless
        // we're running with no delay between events, in which case set a minimal buffer as we're
        // trying to cause clients to get ejected for lagging.
        let config = Config {
            event_stream_buffer_length: if server_behavior.has_delay_between_events {
                BUFFER_LENGTH
            } else {
                1
            },
            max_concurrent_subscribers: server_behavior
                .max_concurrent_subscribers
                .unwrap_or(Config::default().max_concurrent_subscribers),
            ..Default::default()
        };
        let mut server = EventStreamServer::new(
            config,
            self.storage_dir.path().to_path_buf(),
            self.protocol_version,
            self.deploy_getter.clone(),
        )
        .unwrap();
        self.first_event_id = server.event_indexer.current_index();
        let first_event_id = server.event_indexer.current_index();
        let server_address = server.listening_address;
        let events = self.events.clone();
        let server_stopper = self.server_stopper.clone();
        let join_handle = tokio::spawn(async move {
            let event_count = if server_behavior.repeat_events {
                MAX_EVENT_COUNT
            } else {
                EVENT_COUNT
            };
            for (id, event) in events.iter().cycle().enumerate().take(event_count as usize) {
                if server_stopper.should_stop() {
                    debug!("stopping server early");
                    return;
                }
                server_behavior
                    .wait_for_clients((id as Id).wrapping_add(first_event_id))
                    .await;
                let _ = server.broadcast(event.clone());
                server_behavior.sleep_if_required().await;
            }
            // Keep the server running until told to stop.  Clients connecting from now will only
            // receive keepalives.
            debug!("server finished sending all events");
            server_stopper.wait().await;
            debug!("server stopped");
        });
        self.server_join_handle = Some(join_handle);
        server_address
    }
async fn run_server ( & mut self , server_behavior : ServerBehavior ) -> SocketAddr { if self . server_join_handle . is_some ( ) { panic ! ( " " ) ; } self . server_stopper = ServerStopper :: new ( ) ; // Set the server to use a channel buffer of half the total events it will emit, unless // we're running with no delay between events, in which case set a minimal buffer as we're // trying to cause clients to get ejected for lagging. let config = Config { event_stream_buffer_length : if server_behavior . has_delay_between_events { BUFFER_LENGTH } else { 1 } , max_concurrent_subscribers : server_behavior . max_concurrent_subscribers . unwrap_or ( Config :: default ( ) . max_concurrent_subscribers ) , .. Default :: default ( ) } ; let mut server = EventStreamServer :: new ( config , self . storage_dir . path ( ) . to_path_buf ( ) , self . protocol_version , self . deploy_getter . clone ( ) , ) . unwrap ( ) ; self . first_event_id = server . event_indexer . current_index ( ) ; let first_event_id = server . event_indexer . current_index ( ) ; let server_address = server . listening_address ; let events = self . events . clone ( ) ; let server_stopper = self . server_stopper . clone ( ) ; let join_handle = tokio :: spawn ( async move { let event_count = if server_behavior . repeat_events { MAX_EVENT_COUNT } else { EVENT_COUNT } ; for ( id , event ) in events . iter ( ) . cycle ( ) . enumerate ( ) . take ( event_count as usize ) { if server_stopper . should_stop ( ) { debug ! ( " " ) ; return ; } server_behavior . wait_for_clients ( ( id as Id ) . wrapping_add ( first_event_id ) ) . await ; let _ = server . broadcast ( event . clone ( ) ) ; server_behavior . sleep_if_required ( ) . await ; } // Keep the server running until told to stop.  Clients connecting from now will only // receive keepalives. debug ! ( " " ) ; server_stopper . wait ( ) . await ; debug ! ( " " ) ; } ) ; self . server_join_handle = Some ( join_handle ) ; server_address }
Creates a new `EventStreamServer` and runs it in a tokio task, returning the actual address
the server is listening on.
Creates a new ` EventStreamServer ` and runs it in a tokio task returning the actual address the server is listening on .
&mut self
server_behavior
ServerBehavior
&mut self
server_behavior
ServerBehavior