Struct coordtest::CoordTest

pub struct CoordTest {
    pub(crate) dataflow_client: InterceptingDataflowClient<LocalClient>,
    pub(crate) coord_client: Client,
    pub(crate) _coord_handle: Handle,
    pub(crate) _dataflow_server: Server,
    pub(crate) queued_feedback: Vec<Response>,
    pub(crate) _data_directory: TempDir,
    pub(crate) temp_dir: TempDir,
    pub(crate) uppers: HashMap<GlobalId, Timestamp>,
    pub(crate) timestamp: Arc<Mutex<u64>>,
    pub(crate) verbose: bool,
    pub(crate) persisted_sessions: HashMap<String, (SessionClient, StartupResponse)>,
    pub(crate) deferred_results: HashMap<String, Vec<ExecuteResponse>>,
}

CoordTest works by creating a Coordinator with mechanisms to control when it receives messages. The dataflow server is started with a single worker, but it’s feedback channel is controlled by the InterceptingDataflowClient, allowing us to control when uppers and sinces advance.

Fields

dataflow_client: InterceptingDataflowClient<LocalClient>

coord_client: Client

_coord_handle: Handle

_dataflow_server: Server

queued_feedback: Vec<Response>

_data_directory: TempDir

temp_dir: TempDir

uppers: HashMap<GlobalId, Timestamp>

timestamp: Arc<Mutex<u64>>

verbose: bool

persisted_sessions: HashMap<String, (SessionClient, StartupResponse)>

deferred_results: HashMap<String, Vec<ExecuteResponse>>

Implementations

impl CoordTest

pub async fn new() -> Result<Self>

pub(crate) async fn connect(&self) -> Result<(SessionClient, StartupResponse)>

pub(crate) fn rewrite_query(&self, query: &str) -> String

pub(crate) async fn with_sc_inner<F, T>(
    &mut self,
    session_name: Option<String>,
    f: F
) -> T where
    F: for<'a> FnOnce(&'a mut SessionClient) -> Pin<Box<dyn Future<Output = T> + 'a>>, 

pub(crate) async fn with_sc<F, T>(&mut self, f: F) -> T where
    F: for<'a> FnOnce(&'a mut SessionClient) -> Pin<Box<dyn Future<Output = T> + 'a>>, 

Provide a SessionClient to f, but terminate the client after executing f.

pub(crate) async fn with_persisted_sc<F, T>(
    &mut self,
    session_name: String,
    f: F
) -> T where
    F: for<'a> FnOnce(&'a mut SessionClient) -> Pin<Box<dyn Future<Output = T> + 'a>>, 

Rather than terminating the SessionClient after executing f, store it on self using some name.

pub(crate) async fn drain_feedback_msgs(&mut self)

pub(crate) async fn drain_skip_uppers(
    &mut self,
    exclude_uppers: &HashSet<GlobalId>
)

pub(crate) async fn wait_for_peek(
    &mut self,
    rows: Pin<Box<dyn Future<Output = PeekResponse>>>
) -> PeekResponse

pub(crate) async fn drain_peek_response(&mut self)

pub(crate) async fn make_catalog(&mut self) -> Catalog