text:
use crate::{
    driver::DriverConfiguration,
    error::Error,
    logging::{LogEntry, LogSchema},
    metadata_storage::MetadataStorageInterface,
    storage_synchronizer::StorageSynchronizerInterface,
    utils,
    utils::{SpeculativeStreamState, PENDING_DATA_LOG_FREQ_SECS},
};
use aptos_config::config::BootstrappingMode;
use aptos_data_client::GlobalDataSummary;
use aptos_logger::{prelude::*, sample::SampleRate};
use aptos_types::{
    epoch_change::Verifier,
    epoch_state::EpochState,
    ledger_info::LedgerInfoWithSignatures,
    state_store::state_value::StateValueChunkWithProof,
    transaction::{TransactionListWithProof, TransactionOutputListWithProof, Version},
    waypoint::Waypoint,
};
use data_streaming_service::{
    data_notification::{DataNotification, DataPayload, NotificationId},
    data_stream::DataStreamListener,
    streaming_client::{DataStreamingClient, NotificationAndFeedback, NotificationFeedback},
};
use futures::channel::oneshot;
use std::{collections::BTreeMap, sync::Arc, time::Duration};
use storage_interface::DbReader;

/// The expected version of the genesis transaction
pub const GENESIS_TRANSACTION_VERSION: u64 = 0;

/// A simple container for verified epoch states and epoch ending ledger infos
/// that have been fetched from the network.
pub(crate) struct VerifiedEpochStates {
    // If new epoch ending ledger infos have been fetched from the network
    fetched_epoch_ending_ledger_infos: bool,

    // The highest epoch ending version fetched thus far
    highest_fetched_epoch_ending_version: Version,

    // The latest epoch state that has been verified by the node
    latest_epoch_state: EpochState,

    // A map from versions to epoch ending ledger infos fetched from the network
    new_epoch_ending_ledger_infos: BTreeMap<Version, LedgerInfoWithSignatures>,

    // If the node has successfully verified the waypoint
    verified_waypoint: bool,
}

impl VerifiedEpochStates {
    pub fn new(latest_epoch_state: EpochState) -> Self {
        Self {
            fetched_epoch_ending_ledger_infos: false,
            highest_fetched_epoch_ending_version: 0,
            latest_epoch_state,
            new_epoch_ending_ledger_infos: BTreeMap::new(),
            verified_waypoint: false,
        }
    }

    /// Returns true iff the node has already fetched any new epoch
    /// ending ledger infos from the network.
    pub fn fetched_epoch_ending_ledger_infos(&self) -> bool {
        self.fetched_epoch_ending_ledger_infos
    }

    /// Sets `fetched_epoch_ending_ledger_infos` to true
    pub fn set_fetched_epoch_ending_ledger_infos(&mut self) {
        self.fetched_epoch_ending_ledger_infos = true;
    }

    /// Returns true iff the node has verified the waypoint
    pub fn verified_waypoint(&self) -> bool {
        self.verified_waypoint
    }

    /// Sets `verified_waypoint` to true
    pub fn set_verified_waypoint(&mut self) {
        self.verified_waypoint = true;
    }

    /// Verifies the given epoch ending ledger info, updates our latest
    /// trusted epoch state and attempts to verify any given waypoint.
    pub fn update_verified_epoch_states(
        &mut self,
        epoch_ending_ledger_info: &LedgerInfoWithSignatures,
        waypoint: &Waypoint,
    ) -> Result<(), Error> {
        // Verify the ledger info against the latest epoch state
        self.latest_epoch_state
            .verify(epoch_ending_ledger_info)
            .map_err(|error| {
                Error::VerificationError(format!("Ledger info failed verification: {:?}", error))
            })?;

        // Update the latest epoch state with the next epoch
        if let Some(next_epoch_state) = epoch_ending_ledger_info.ledger_info().next_epoch_state() {
            self.highest_fetched_epoch_ending_version =
                epoch_ending_ledger_info.ledger_info().version();
            self.latest_epoch_state = next_epoch_state.clone();
            self.insert_new_epoch_ending_ledger_info(epoch_ending_ledger_info.clone());

            trace!(LogSchema::new(LogEntry::Bootstrapper).message(&format!(
                "Updated the latest epoch state to epoch: {:?}",
                self.latest_epoch_state.epoch
            )));
        } else {
            return Err(Error::VerificationError(
                "The ledger info was not epoch ending!".into(),
            ));
        }

        // Check if the ledger info corresponds to the trusted waypoint
        self.verify_waypoint(epoch_ending_ledger_info, waypoint)
    }

    /// Attempts to verify the waypoint using the new epoch ending ledger info
    fn verify_waypoint(
        &mut self,
        epoch_ending_ledger_info: &LedgerInfoWithSignatures,
        waypoint: &Waypoint,
    ) -> Result<(), Error> {
        if !self.verified_waypoint {
            // Fetch the waypoint and ledger info versions
            let waypoint_version = waypoint.version();
            let ledger_info = epoch_ending_ledger_info.ledger_info();
            let ledger_info_version = ledger_info.version();

            // Verify we haven't missed the waypoint
            if ledger_info_version > waypoint_version {
                return Err(Error::VerificationError(
                    format!("Failed to verify the waypoint: ledger info version is too high! Waypoint version: {:?}, ledger info version: {:?}",
                            waypoint_version, ledger_info_version)
                ));
            }

            // Check if we've found the ledger info corresponding to the waypoint version
            if ledger_info_version == waypoint_version {
                match waypoint.verify(ledger_info) {
                    Ok(()) => self.set_verified_waypoint(),
                    Err(error) => {
                        return Err(Error::VerificationError(
                            format!("Failed to verify the waypoint: {:?}! Waypoint: {:?}, given ledger info: {:?}",
                                    error, waypoint, ledger_info)
                        ));
                    }
                }
            }
        }

        Ok(())
    }

    /// Adds an epoch ending ledger info to the new epoch ending ledger infos map
    fn insert_new_epoch_ending_ledger_info(
        &mut self,
        epoch_ending_ledger_info: LedgerInfoWithSignatures,
    ) {
        let ledger_info = epoch_ending_ledger_info.ledger_info();
        info!(LogSchema::new(LogEntry::Bootstrapper).message(&format!(
            "Adding a new epoch to the epoch ending ledger infos. Epoch: {:?}, Version: {:?}, Ends epoch: {:?}",
            ledger_info.epoch(), ledger_info.version(), ledger_info.ends_epoch(),
        )));

        // Insert the version to ledger info mapping
        let version = ledger_info.version();
        if let Some(epoch_ending_ledger_info) = self
            .new_epoch_ending_ledger_infos
            .insert(version, epoch_ending_ledger_info)
        {
            panic!(
                "Duplicate epoch ending ledger info found!\
                 Version: {:?}, \
                 ledger info: {:?}",
                version, epoch_ending_ledger_info,
            );
        }
    }

    /// Return all epoch ending ledger infos
    pub fn all_epoch_ending_ledger_infos(&self) -> Vec<LedgerInfoWithSignatures> {
        self.new_epoch_ending_ledger_infos
            .values()
            .cloned()
            .collect()
    }

    /// Returns any epoch ending ledger info associated with the given version
    pub fn get_epoch_ending_ledger_info(
        &self,
        version: Version,
    ) -> Option<LedgerInfoWithSignatures> {
        self.new_epoch_ending_ledger_infos.get(&version).cloned()
    }

    /// Returns the highest known ledger info we've fetched (if any)
    pub fn get_highest_known_ledger_info(&self) -> Option<LedgerInfoWithSignatures> {
        if !self.new_epoch_ending_ledger_infos.is_empty() {
            let highest_fetched_ledger_info = self
                .get_epoch_ending_ledger_info(self.highest_fetched_epoch_ending_version)
                .unwrap_or_else(|| {
                    panic!(
                        "The highest known ledger info for version: {:?} was not found!",
                        self.highest_fetched_epoch_ending_version
                    )
                });
            Some(highest_fetched_ledger_info)
        } else {
            None
        }
    }

    /// Returns the next epoch ending version after the given version (if one
    /// exists).
    pub fn next_epoch_ending_version(&self, version: Version) -> Option<Version> {
        // BTreeMap keys are iterated through in increasing key orders (i.e., versions)
        for (epoch_ending_version, _) in self.new_epoch_ending_ledger_infos.iter() {
            if *epoch_ending_version > version {
                return Some(*epoch_ending_version);
            }
        }
        None
    }
}

/// A simple container to manage data related to state value snapshot syncing
pub(crate) struct StateValueSyncer {
    // Whether or not a state snapshot receiver has been initialized
    initialized_state_snapshot_receiver: bool,

    // The epoch ending ledger info for the version we're syncing
    ledger_info_to_sync: Option<LedgerInfoWithSignatures>,

    // The next state value index to process (all state values before this have been
    // processed -- i.e., sent to the storage synchronizer).
    next_state_index_to_process: u64,

    // The transaction output (inc. info and proof) for the version we're syncing
    transaction_output_to_sync: Option<TransactionOutputListWithProof>,
}

impl StateValueSyncer {
    pub fn new() -> Self {
        Self {
            initialized_state_snapshot_receiver: false,
            ledger_info_to_sync: None,
            next_state_index_to_process: 0,
            transaction_output_to_sync: None,
        }
    }

    /// Sets the ledger info to sync
    pub fn set_ledger_info_to_sync(&mut self, ledger_info_to_sync: LedgerInfoWithSignatures) {
        self.ledger_info_to_sync = Some(ledger_info_to_sync);
    }

    /// Sets the transaction output to sync
    pub fn set_transaction_output_to_sync(
        &mut self,
        transaction_output_to_sync: TransactionOutputListWithProof,
    ) {
        self.transaction_output_to_sync = Some(transaction_output_to_sync);
    }

    /// Updates the next state index to process
    pub fn update_next_state_index_to_process(&mut self, next_state_index_to_process: u64) {
        self.next_state_index_to_process = next_state_index_to_process;
    }
}

/// A simple component that manages the bootstrapping of the node
pub struct Bootstrapper<MetadataStorage, StorageSyncer, StreamingClient> {
    // The currently active data stream (provided by the data streaming service)
    active_data_stream: Option<DataStreamListener>,

    // The channel used to notify a listener of successful bootstrapping
    bootstrap_notifier_channel: Option<oneshot::Sender<Result<(), Error>>>,

    // If the node has completed bootstrapping
    bootstrapped: bool,

    // The config of the state sync driver
    driver_configuration: DriverConfiguration,

    // The storage to write metadata about the syncing progress
    metadata_storage: MetadataStorage,

    // The speculative state tracking the active data stream
    speculative_stream_state: Option<SpeculativeStreamState>,

    // The component used to sync state values (if downloading states)
    state_value_syncer: StateValueSyncer,

    // The client through which to stream data from the Aptos network
    streaming_client: StreamingClient,

    // The interface to read from storage
    storage: Arc<dyn DbReader>,

    // The storage synchronizer used to update local storage
    storage_synchronizer: StorageSyncer,

    // The epoch states verified by this node (held in memory)
    verified_epoch_states: VerifiedEpochStates,
}

impl<
        MetadataStorage: MetadataStorageInterface + Clone,
        StorageSyncer: StorageSynchronizerInterface + Clone,
        StreamingClient: DataStreamingClient + Clone,
    > Bootstrapper<MetadataStorage, StorageSyncer, StreamingClient>
{
    pub fn new(
        driver_configuration: DriverConfiguration,
        metadata_storage: MetadataStorage,
        streaming_client: StreamingClient,
        storage: Arc<dyn DbReader>,
        storage_synchronizer: StorageSyncer,
    ) -> Self {
        // Load the latest epoch state from storage
        let latest_epoch_state = utils::fetch_latest_epoch_state(storage.clone())
            .expect("Unable to fetch latest epoch state!");
        let verified_epoch_states = VerifiedEpochStates::new(latest_epoch_state);

        Self {
            state_value_syncer: StateValueSyncer::new(),
            active_data_stream: None,
            bootstrap_notifier_channel: None,
            bootstrapped: false,
            driver_configuration,
            metadata_storage,
            speculative_stream_state: None,
            streaming_client,
            storage,
            storage_synchronizer,
            verified_epoch_states,
        }
    }

    /// Returns true iff the node has already completed bootstrapping
    pub fn is_bootstrapped(&self) -> bool {
        self.bootstrapped
    }

    /// Marks bootstrapping as complete and notifies any listeners
    pub async fn bootstrapping_complete(&mut self) -> Result<(), Error> {
        info!(LogSchema::new(LogEntry::Bootstrapper)
            .message("The node has successfully bootstrapped!"));
        self.bootstrapped = true;
        self.notify_listeners_if_bootstrapped().await
    }

    /// Subscribes the specified channel to bootstrap completion notifications
    pub async fn subscribe_to_bootstrap_notifications(
        &mut self,
        bootstrap_notifier_channel: oneshot::Sender<Result<(), Error>>,
    ) -> Result<(), Error> {
        if self.bootstrap_notifier_channel.is_some() {
            panic!("Only one boostrap subscriber is supported at a time!");
        }

        self.bootstrap_notifier_channel = Some(bootstrap_notifier_channel);
        self.notify_listeners_if_bootstrapped().await
    }

    /// Notifies any listeners if we've now bootstrapped
    async fn notify_listeners_if_bootstrapped(&mut self) -> Result<(), Error> {
        if self.is_bootstrapped() {
            if let Some(notifier_channel) = self.bootstrap_notifier_channel.take() {
                if let Err(error) = notifier_channel.send(Ok(())) {
                    return Err(Error::CallbackSendFailed(format!(
                        "Bootstrap notification error: {:?}",
                        error
                    )));
                }
            }
            self.reset_active_stream(None).await?;
            self.storage_synchronizer.finish_chunk_executor(); // The bootstrapper is now complete
        }

        Ok(())
    }

    /// Checks if the bootstrapper is able to make progress
    pub async fn drive_progress(
        &mut self,
        global_data_summary: &GlobalDataSummary,
    ) -> Result<(), Error> {
        if self.is_bootstrapped() {
            return Err(Error::AlreadyBootstrapped(
                "The bootstrapper should not attempt to make progress!".into(),
            ));
        }

        if self.active_data_stream.is_some() {
            // We have an active data stream. Process any notifications!
            self.process_active_stream_notifications().await?;
        } else if self.storage_synchronizer.pending_storage_data() {
            // Wait for any pending data to be processed
            sample!(
                SampleRate::Duration(Duration::from_secs(PENDING_DATA_LOG_FREQ_SECS)),
                info!("Waiting for the storage synchronizer to handle pending data!")
            );
        } else {
            // Fetch a new data stream to start streaming data
            self.initialize_active_data_stream(global_data_summary)
                .await?;
        }

        // Check if we've now bootstrapped
        self.notify_listeners_if_bootstrapped().await
    }

    /// Returns true iff the bootstrapper should continue to fetch epoch ending
    /// ledger infos (in order to make progress).
    fn should_fetch_epoch_ending_ledger_infos(&self) -> bool {
        !self
            .verified_epoch_states
            .fetched_epoch_ending_ledger_infos()
            || !self.verified_epoch_states.verified_waypoint()
    }

    /// Initializes an active data stream so that we can begin to process notifications
    async fn initialize_active_data_stream(
        &mut self,
        global_data_summary: &GlobalDataSummary,
    ) -> Result<(), Error> {
        // Reset the chunk executor to flush any invalid state currently held in-memory
        self.storage_synchronizer.reset_chunk_executor()?;

        // Always fetch the new epoch ending ledger infos first
        if self.should_fetch_epoch_ending_ledger_infos() {
            return self
                .fetch_epoch_ending_ledger_infos(global_data_summary)
                .await;
        }

        // Get the highest synced and known ledger info versions
        let highest_synced_version = utils::fetch_latest_synced_version(self.storage.clone())?;
        let highest_known_ledger_info = self.get_highest_known_ledger_info()?;
        let highest_known_ledger_version = highest_known_ledger_info.ledger_info().version();

        // If we've already synced to the highest known version, there's nothing to do
        if highest_synced_version >= highest_known_ledger_version {
            info!(LogSchema::new(LogEntry::Bootstrapper)
                .message(&format!("Highest synced version {} is >= highest known ledger version {}, nothing needs to be done.",
                    highest_synced_version, highest_known_ledger_version)));
            return self.bootstrapping_complete().await;
        }

        info!(LogSchema::new(LogEntry::Bootstrapper).message(&format!(
            "Highest synced version is {}, highest_known_ledger_info is {:?}, bootstrapping_mode is {:?}.",
            highest_synced_version, highest_known_ledger_info,
            self.driver_configuration.config.bootstrapping_mode)));

        // Bootstrap according to the mode
        match self.driver_configuration.config.bootstrapping_mode {
            BootstrappingMode::DownloadLatestStates => {
                self.fetch_missing_state_snapshot_data(
                    highest_synced_version,
                    highest_known_ledger_info,
                )
                .await
            }
            _ => {
                // We're either transaction or output syncing
                self.fetch_missing_transaction_data(
                    highest_synced_version,
                    highest_known_ledger_info,
                )
                .await
            }
        }
    }

    /// Fetches all missing state snapshot data in order to bootstrap the node
    async fn fetch_missing_state_snapshot_data(
        &mut self,
        highest_synced_version: Version,
        highest_known_ledger_info: LedgerInfoWithSignatures,
    ) -> Result<(), Error> {
        if highest_synced_version == GENESIS_TRANSACTION_VERSION {
            // We're syncing a new node. Check the progress and fetch the missing data.
            if let Some(target) = self.metadata_storage.previous_snapshot_sync_target()? {
                if self.metadata_storage.is_snapshot_sync_complete(&target)? {
                    panic!(
                        "The snapshot sync for the target was marked as complete but \
                    the highest synced version is genesis! Something has gone wrong! \
                    Target snapshot sync: {:?}",
                        target
                    );
                }
                self.fetch_missing_state_values(target, true).await
            } else {
                // No snapshot sync has started. Start a new sync for the highest known ledger info.
                self.fetch_missing_state_values(highest_known_ledger_info, false)
                    .await
            }
        } else {
            // This node has already synced some state. Ensure the node is not too far behind.
            let highest_known_ledger_version = highest_known_ledger_info.ledger_info().version();
            let num_versions_behind = highest_known_ledger_version
                .checked_sub(highest_synced_version)
                .ok_or_else(|| {
                    Error::IntegerOverflow("The number of versions behind has overflown!".into())
                })?;
            if num_versions_behind
                < self
                    .driver_configuration
                    .config
                    .num_versions_to_skip_snapshot_sync
            {
                info!(LogSchema::new(LogEntry::Bootstrapper).message(&format!(
                    "The node is only {} versions behind, will skip bootstrapping.",
                    num_versions_behind
                )));
                // We've already bootstrapped to an initial state snapshot. If this a fullnode, the
                // continuous syncer will take control and get the node up-to-date. If this is a
                // validator, consensus will take control and sync depending on how it sees fit.
                self.bootstrapping_complete().await
            } else {
                panic!("Fast syncing is currently unsupported for nodes with existing state! \
                        You are currently {:?} versions behind the latest snapshot version ({:?}). Either \
                        select a different syncing mode, or delete your storage and restart your node.",
                       num_versions_behind, highest_known_ledger_version);
            }
        }
    }

    /// Attempts to fetch a data notification from the active stream
    async fn fetch_next_data_notification(&mut self) -> Result<DataNotification, Error> {
        let max_stream_wait_time_ms = self.driver_configuration.config.max_stream_wait_time_ms;
        let max_num_stream_timeouts = self.driver_configuration.config.max_num_stream_timeouts;
        let result = utils::get_data_notification(
            max_stream_wait_time_ms,
            max_num_stream_timeouts,
            self.active_data_stream.as_mut(),
        )
        .await;
        if matches!(result, Err(Error::CriticalDataStreamTimeout(_))) {
            // If the stream has timed out too many times, we need to reset it
            warn!("Resetting the currently active data stream due to too many timeouts!");
            self.reset_active_stream(None).await?;
        }
        result
    }

    /// Processes any notifications already pending on the active stream
    async fn process_active_stream_notifications(&mut self) -> Result<(), Error> {
        for _ in 0..self
            .driver_configuration
            .config
            .max_consecutive_stream_notifications
        {
            // Fetch and process any data notifications
            let data_notification = self.fetch_next_data_notification().await?;
            match data_notification.data_payload {
                DataPayload::StateValuesWithProof(state_value_chunk_with_proof) => {
                    self.process_state_values_payload(
                        data_notification.notification_id,
                        state_value_chunk_with_proof,
                    )
                    .await?;
                }
                DataPayload::EpochEndingLedgerInfos(epoch_ending_ledger_infos) => {
                    self.process_epoch_ending_payload(
                        data_notification.notification_id,
                        epoch_ending_ledger_infos,
                    )
                    .await?;
                }
                DataPayload::TransactionsWithProof(transactions_with_proof) => {
                    let payload_start_version = transactions_with_proof.first_transaction_version;
                    self.process_transaction_or_output_payload(
                        data_notification.notification_id,
                        Some(transactions_with_proof),
                        None,
                        payload_start_version,
                    )
                    .await?;
                }
                DataPayload::TransactionOutputsWithProof(transaction_outputs_with_proof) => {
                    let payload_start_version =
                        transaction_outputs_with_proof.first_transaction_output_version;
                    self.process_transaction_or_output_payload(
                        data_notification.notification_id,
                        None,
                        Some(transaction_outputs_with_proof),
                        payload_start_version,
                    )
                    .await?;
                }
                _ => {
                    return self
                        .handle_end_of_stream_or_invalid_payload(data_notification)
                        .await
                }
            }
        }

        Ok(())
    }

    /// Fetches state values (as required to bootstrap the node)
    async fn fetch_missing_state_values(
        &mut self,
        target_ledger_info: LedgerInfoWithSignatures,
        existing_snapshot_progress: bool,
    ) -> Result<(), Error> {
        // Initialize the target ledger info and verify it never changes
        if let Some(ledger_info_to_sync) = &self.state_value_syncer.ledger_info_to_sync {
            if ledger_info_to_sync != &target_ledger_info {
                panic!(
                    "Mismatch in ledger info to sync! Given target: {:?}, stored target: {:?}",
                    target_ledger_info, ledger_info_to_sync
                );
            }
        } else {
            self.state_value_syncer
                .set_ledger_info_to_sync(target_ledger_info.clone());
        }

        // Fetch the data that we're missing
        let target_ledger_info_version = target_ledger_info.ledger_info().version();
        let data_stream = if self.state_value_syncer.transaction_output_to_sync.is_none() {
            // Fetch the transaction info first, before the states
            self.streaming_client
                .get_all_transaction_outputs(
                    target_ledger_info_version,
                    target_ledger_info_version,
                    target_ledger_info_version,
                )
                .await?
        } else {
            // Identify the next state index to fetch
            let next_state_index_to_process = if existing_snapshot_progress {
                // The state snapshot receiver requires that after each reboot we
                // rewrite the last persisted index (again!). This is a limitation
                // of how the snapshot is persisted (i.e., in-memory sibling freezing).
                // Thus, on each stream reset, we overlap every chunk by a single item.
                self
                    .metadata_storage
                    .get_last_persisted_state_value_index(&target_ledger_info)
                    .map_err(|error| {
                        Error::StorageError(format!(
                            "Failed to get the last persisted state value index at version {:?}! Error: {:?}",
                            target_ledger_info_version, error
                        ))
                    })?
            } else {
                0 // We need to start the snapshot sync from index 0
            };

            // Fetch the missing state values
            self.state_value_syncer
                .update_next_state_index_to_process(next_state_index_to_process);
            self.streaming_client
                .get_all_state_values(
                    target_ledger_info_version,
                    Some(next_state_index_to_process),
                )
                .await?
        };
        self.active_data_stream = Some(data_stream);

        Ok(())
    }

    /// Fetches all missing transaction data in order to bootstrap the node
    async fn fetch_missing_transaction_data(
        &mut self,
        highest_synced_version: Version,
        highest_known_ledger_info: LedgerInfoWithSignatures,
    ) -> Result<(), Error> {
        let highest_known_ledger_version = highest_known_ledger_info.ledger_info().version();
        let next_version = highest_synced_version.checked_add(1).ok_or_else(|| {
            Error::IntegerOverflow("The next output version has overflown!".into())
        })?;
        let end_version = self
            .verified_epoch_states
            .next_epoch_ending_version(highest_synced_version)
            .expect("No higher epoch ending version known!");
        let data_stream = match self.driver_configuration.config.bootstrapping_mode {
            BootstrappingMode::ApplyTransactionOutputsFromGenesis => {
                self.streaming_client
                    .get_all_transaction_outputs(
                        next_version,
                        end_version,
                        highest_known_ledger_version,
                    )
                    .await?
            }
            BootstrappingMode::ExecuteTransactionsFromGenesis => {
                self.streaming_client
                    .get_all_transactions(
                        next_version,
                        end_version,
                        highest_known_ledger_version,
                        false,
                    )
                    .await?
            }
            bootstrapping_mode => {
                unreachable!("Bootstrapping mode not supported: {:?}", bootstrapping_mode)
            }
        };
        self.speculative_stream_state = Some(SpeculativeStreamState::new(
            utils::fetch_latest_epoch_state(self.storage.clone())?,
            Some(highest_known_ledger_info),
            highest_synced_version,
        ));
        self.active_data_stream = Some(data_stream);

        Ok(())
    }

    /// Fetches all epoch ending ledger infos (from the current epoch to the
    /// maximum that can be found by the data streaming service).
    async fn fetch_epoch_ending_ledger_infos(
        &mut self,
        global_data_summary: &GlobalDataSummary,
    ) -> Result<(), Error> {
        // Verify the waypoint can be satisfied
        self.verify_waypoint_is_satisfiable(global_data_summary)?;

        // Get the highest advertised epoch that has ended
        let highest_advertised_epoch_end = global_data_summary
            .advertised_data
            .highest_epoch_ending_ledger_info()
            .ok_or_else(|| {
                Error::AdvertisedDataError(
                    "No highest advertised epoch end found in the network!".into(),
                )
            })?;

        // Fetch the highest epoch end known locally
        let highest_known_ledger_info = self.get_highest_known_ledger_info()?;
        let highest_known_ledger_info = highest_known_ledger_info.ledger_info();
        let highest_local_epoch_end = if highest_known_ledger_info.ends_epoch() {
            highest_known_ledger_info.epoch()
        } else if highest_known_ledger_info.epoch() > 0 {
            highest_known_ledger_info
                .epoch()
                .checked_sub(1)
                .ok_or_else(|| {
                    Error::IntegerOverflow("The highest local epoch end has overflown!".into())
                })?
        } else {
            unreachable!("Genesis should always end the first epoch!");
        };

        // Compare the highest local epoch end to the highest advertised epoch end
        if highest_local_epoch_end > highest_advertised_epoch_end {
            let error_message =
                format!(
                    "The highest local epoch end is higher than the advertised epoch end! Local: {:?}, advertised: {:?}",
                    highest_local_epoch_end, highest_advertised_epoch_end
                );
            return Err(Error::AdvertisedDataError(error_message));
        } else if highest_local_epoch_end < highest_advertised_epoch_end {
            info!(LogSchema::new(LogEntry::Bootstrapper).message(&format!(
                "Found higher epoch ending ledger infos in the network! Local: {:?}, advertised: {:?}",
                   highest_local_epoch_end, highest_advertised_epoch_end
            )));
            let next_epoch_end = highest_local_epoch_end.checked_add(1).ok_or_else(|| {
                Error::IntegerOverflow("The next epoch end has overflown!".into())
            })?;
            let epoch_ending_stream = self
                .streaming_client
                .get_all_epoch_ending_ledger_infos(next_epoch_end)
                .await?;
            self.active_data_stream = Some(epoch_ending_stream);
        } else if self.verified_epoch_states.verified_waypoint() {
            info!(LogSchema::new(LogEntry::Bootstrapper).message(
                "No new epoch ending ledger infos to fetch! All peers are in the same epoch!"
            ));
            self.verified_epoch_states
                .set_fetched_epoch_ending_ledger_infos();
        } else {
            return Err(Error::AdvertisedDataError("Our waypoint is unverified, but there's no higher epoch ending ledger infos advertised!".into()));
        };

        Ok(())
    }

    /// Verifies that connected peers have advertised data beyond our waypoint
    /// or that our waypoint is trivially satisfiable.
    fn verify_waypoint_is_satisfiable(
        &mut self,
        global_data_summary: &GlobalDataSummary,
    ) -> Result<(), Error> {
        // If our storage has already synced beyond our waypoint, nothing needs to be checked
        let latest_ledger_info = utils::fetch_latest_synced_ledger_info(self.storage.clone())?;
        let waypoint_version = self.driver_configuration.waypoint.version();
        if latest_ledger_info.ledger_info().version() >= waypoint_version {
            self.verified_epoch_states.set_verified_waypoint();
            return Ok(());
        }

        // Get the highest advertised synced ledger info version
        let highest_advertised_ledger_info = global_data_summary
            .advertised_data
            .highest_synced_ledger_info()
            .ok_or_else(|| {
                Error::AdvertisedDataError(
                    "No highest advertised ledger info found in the network!".into(),
                )
            })?;
        let highest_advertised_version = highest_advertised_ledger_info.ledger_info().version();

        // Compare the highest advertised version with our waypoint
        if highest_advertised_version < waypoint_version {
            Err(Error::AdvertisedDataError(
                format!(
                    "No advertised version higher than our waypoint! Highest version: {:?}, waypoint version: {:?}",
                    highest_advertised_version, waypoint_version
                )
            ))
        } else {
            Ok(())
        }
    }

    /// Verifies the start and end indices in the given state value chunk
    async fn verify_states_values_indices(
        &mut self,
        notification_id: NotificationId,
        state_value_chunk_with_proof: &StateValueChunkWithProof,
    ) -> Result<(), Error> {
        // Verify the payload start index is valid
        let expected_start_index = self.state_value_syncer.next_state_index_to_process;
        if expected_start_index != state_value_chunk_with_proof.first_index {
            self.reset_active_stream(Some(NotificationAndFeedback::new(
                notification_id,
                NotificationFeedback::InvalidPayloadData,
            )))
            .await?;
            return Err(Error::VerificationError(format!(
                "The start index of the state values was invalid! Expected: {:?}, received: {:?}",
                expected_start_index, state_value_chunk_with_proof.first_index
            )));
        }

        // Verify the end index and number of state values is valid
        let expected_num_state_values = state_value_chunk_with_proof
            .last_index
            .checked_sub(state_value_chunk_with_proof.first_index)
            .and_then(|version| version.checked_add(1)) // expected_num_state_values = last_index - first_index + 1
            .ok_or_else(|| {
                Error::IntegerOverflow("The expected number of state values has overflown!".into())
            })?;
        let num_state_values = state_value_chunk_with_proof.raw_values.len() as u64;
        if expected_num_state_values != num_state_values {
            self.reset_active_stream(Some(NotificationAndFeedback::new(
                notification_id,
                NotificationFeedback::InvalidPayloadData,
            )))
            .await?;
            return Err(Error::VerificationError(format!(
                "The expected number of state values was invalid! Expected: {:?}, received: {:?}",
                expected_num_state_values, num_state_values,
            )));
        }

        Ok(())
    }

    /// Process a single state value chunk with proof payload
    async fn process_state_values_payload(
        &mut self,
        notification_id: NotificationId,
        state_value_chunk_with_proof: StateValueChunkWithProof,
    ) -> Result<(), Error> {
        // Verify that we're expecting state value payloads
        let bootstrapping_mode = self.driver_configuration.config.bootstrapping_mode;
        if self.should_fetch_epoch_ending_ledger_infos()
            || !matches!(bootstrapping_mode, BootstrappingMode::DownloadLatestStates)
        {
            self.reset_active_stream(Some(NotificationAndFeedback::new(
                notification_id,
                NotificationFeedback::InvalidPayloadData,
            )))
            .await?;
            return Err(Error::InvalidPayload(
                "Received an unexpected state values payload!".into(),
            ));
        }

        // Fetch the target ledger info and transaction info for bootstrapping
        let ledger_info_to_sync = self
            .state_value_syncer
            .ledger_info_to_sync
            .clone()
            .expect("Ledger info to sync is missing!");
        let transaction_output_to_sync = self
            .state_value_syncer
            .transaction_output_to_sync
            .clone()
            .expect("Transaction output to sync is missing!");

        // Initialize the state value synchronizer (if not already done)
        if !self.state_value_syncer.initialized_state_snapshot_receiver {
            // Fetch all verified epoch change proofs
            let epoch_change_proofs = self.verified_epoch_states.all_epoch_ending_ledger_infos();

            // Initialize the state value synchronizer
            let _join_handle = self.storage_synchronizer.initialize_state_synchronizer(
                epoch_change_proofs,
                ledger_info_to_sync,
                transaction_output_to_sync.clone(),
            )?;
            self.state_value_syncer.initialized_state_snapshot_receiver = true;
        }

        // Verify the state values payload start and end indices
        self.verify_states_values_indices(notification_id, &state_value_chunk_with_proof)
            .await?;

        // Verify the chunk root hash matches the expected root hash
        let expected_root_hash = transaction_output_to_sync
            .proof
            .transaction_infos
            .first()
            .expect("Target transaction info should exist!")
            .ensure_state_checkpoint_hash()
            .expect("Must be at state checkpoint.");
        if state_value_chunk_with_proof.root_hash != expected_root_hash {
            self.reset_active_stream(Some(NotificationAndFeedback::new(
                notification_id,
                NotificationFeedback::InvalidPayloadData,
            )))
            .await?;
            return Err(Error::VerificationError(format!(
                "The states chunk with proof root hash: {:?} didn't match the expected hash: {:?}!",
                state_value_chunk_with_proof.root_hash, expected_root_hash,
            )));
        }

        // Process the state values chunk and proof
        let last_state_value_index = state_value_chunk_with_proof.last_index;
        if let Err(error) = self
            .storage_synchronizer
            .save_state_values(notification_id, state_value_chunk_with_proof)
        {
            self.reset_active_stream(Some(NotificationAndFeedback::new(
                notification_id,
                NotificationFeedback::InvalidPayloadData,
            )))
            .await?;
            return Err(Error::InvalidPayload(format!(
                "The states chunk with proof was invalid! Error: {:?}",
                error,
            )));
        }

        // Update the next state value index to process
        self.state_value_syncer.next_state_index_to_process =
            last_state_value_index.checked_add(1).ok_or_else(|| {
                Error::IntegerOverflow(
                    "The next state value index to process has overflown!".into(),
                )
            })?;

        Ok(())
    }

    /// Process a single epoch ending payload
    async fn process_epoch_ending_payload(
        &mut self,
        notification_id: NotificationId,
        epoch_ending_ledger_infos: Vec<LedgerInfoWithSignatures>,
    ) -> Result<(), Error> {
        // Verify that we're expecting epoch ending ledger info payloads
        if !self.should_fetch_epoch_ending_ledger_infos() {
            self.reset_active_stream(Some(NotificationAndFeedback::new(
                notification_id,
                NotificationFeedback::InvalidPayloadData,
            )))
            .await?;
            return Err(Error::InvalidPayload(
                "Received an unexpected epoch ending payload!".into(),
            ));
        }

        // Verify the payload isn't empty
        if epoch_ending_ledger_infos.is_empty() {
            self.reset_active_stream(Some(NotificationAndFeedback::new(
                notification_id,
                NotificationFeedback::EmptyPayloadData,
            )))
            .await?;
            return Err(Error::VerificationError(
                "The epoch ending payload was empty!".into(),
            ));
        }

        // Verify the epoch change proofs, update our latest epoch state and
        // verify our waypoint.
        for epoch_ending_ledger_info in epoch_ending_ledger_infos {
            if let Err(error) = self.verified_epoch_states.update_verified_epoch_states(
                &epoch_ending_ledger_info,
                &self.driver_configuration.waypoint,
            ) {
                self.reset_active_stream(Some(NotificationAndFeedback::new(
                    notification_id,
                    NotificationFeedback::PayloadProofFailed,
                )))
                .await?;
                return Err(error);
            }
        }

        // TODO(joshlind): do we want to preemptively notify certain components
        // of the new reconfigurations?

        Ok(())
    }

    /// Process a single transaction or transaction output data payload
    async fn process_transaction_or_output_payload(
        &mut self,
        notification_id: NotificationId,
        transaction_list_with_proof: Option<TransactionListWithProof>,
        transaction_outputs_with_proof: Option<TransactionOutputListWithProof>,
        payload_start_version: Option<Version>,
    ) -> Result<(), Error> {
        // Verify that we're expecting transaction or output payloads
        let bootstrapping_mode = self.driver_configuration.config.bootstrapping_mode;
        if self.should_fetch_epoch_ending_ledger_infos()
            || (matches!(bootstrapping_mode, BootstrappingMode::DownloadLatestStates)
                && self.state_value_syncer.transaction_output_to_sync.is_some())
        {
            self.reset_active_stream(Some(NotificationAndFeedback::new(
                notification_id,
                NotificationFeedback::InvalidPayloadData,
            )))
            .await?;
            return Err(Error::InvalidPayload(
                "Received an unexpected transaction or output payload!".into(),
            ));
        }

        // If we're state syncing, we expect a single transaction info
        if matches!(bootstrapping_mode, BootstrappingMode::DownloadLatestStates) {
            return self
                .verify_transaction_info_to_sync(
                    notification_id,
                    transaction_outputs_with_proof,
                    payload_start_version,
                )
                .await;
        }

        // Verify the payload starting version
        let expected_start_version = self
            .get_speculative_stream_state()
            .expected_next_version()?;
        let payload_start_version = self
            .verify_payload_start_version(
                notification_id,
                payload_start_version,
                expected_start_version,
            )
            .await?;

        // Get the expected proof ledger info for the payload
        let proof_ledger_info = self.get_speculative_stream_state().get_proof_ledger_info();

        // Get the end of epoch ledger info if the payload ends the epoch
        let end_of_epoch_ledger_info = self
            .get_end_of_epoch_ledger_info(
                notification_id,
                payload_start_version,
                transaction_list_with_proof.as_ref(),
                transaction_outputs_with_proof.as_ref(),
            )
            .await?;

        // Execute/apply and commit the transactions/outputs
        let num_transactions_or_outputs = match bootstrapping_mode {
            BootstrappingMode::ApplyTransactionOutputsFromGenesis => {
                if let Some(transaction_outputs_with_proof) = transaction_outputs_with_proof {
                    let num_transaction_outputs = transaction_outputs_with_proof
                        .transactions_and_outputs
                        .len();
                    self.storage_synchronizer
                        .apply_transaction_outputs(
                            notification_id,
                            transaction_outputs_with_proof,
                            proof_ledger_info,
                            end_of_epoch_ledger_info,
                        )
                        .await?;
                    num_transaction_outputs
                } else {
                    self.reset_active_stream(Some(NotificationAndFeedback::new(
                        notification_id,
                        NotificationFeedback::PayloadTypeIsIncorrect,
                    )))
                    .await?;
                    return Err(Error::InvalidPayload(
                        "Did not receive transaction outputs with proof!".into(),
                    ));
                }
            }
            BootstrappingMode::ExecuteTransactionsFromGenesis => {
                if let Some(transaction_list_with_proof) = transaction_list_with_proof {
                    let num_transactions = transaction_list_with_proof.transactions.len();
                    self.storage_synchronizer
                        .execute_transactions(
                            notification_id,
                            transaction_list_with_proof,
                            proof_ledger_info,
                            end_of_epoch_ledger_info,
                        )
                        .await?;
                    num_transactions
                } else {
                    self.reset_active_stream(Some(NotificationAndFeedback::new(
                        notification_id,
                        NotificationFeedback::PayloadTypeIsIncorrect,
                    )))
                    .await?;
                    return Err(Error::InvalidPayload(
                        "Did not receive transactions with proof!".into(),
                    ));
                }
            }
            bootstrapping_mode => {
                unreachable!("Bootstrapping mode not supported: {:?}", bootstrapping_mode)
            }
        };
        let synced_version = payload_start_version
            .checked_add(num_transactions_or_outputs as u64)
            .and_then(|version| version.checked_sub(1)) // synced_version = start + num txns/outputs - 1
            .ok_or_else(|| Error::IntegerOverflow("The synced version has overflown!".into()))?;
        self.get_speculative_stream_state()
            .update_synced_version(synced_version);

        Ok(())
    }

    /// Verifies the payload contains the transaction info we require to
    /// download all state values.
    async fn verify_transaction_info_to_sync(
        &mut self,
        notification_id: NotificationId,
        transaction_outputs_with_proof: Option<TransactionOutputListWithProof>,
        payload_start_version: Option<Version>,
    ) -> Result<(), Error> {
        // Verify the payload starting version
        let ledger_info_to_sync = self
            .state_value_syncer
            .ledger_info_to_sync
            .clone()
            .expect("Ledger info to sync is missing!");
        let expected_start_version = ledger_info_to_sync.ledger_info().version();
        let _ = self
            .verify_payload_start_version(
                notification_id,
                payload_start_version,
                expected_start_version,
            )
            .await?;

        // Verify the payload proof (the ledger info has already been verified)
        // and save the transaction output with proof.
        if let Some(transaction_outputs_with_proof) = transaction_outputs_with_proof {
            if transaction_outputs_with_proof.proof.transaction_infos.len() == 1 {
                match transaction_outputs_with_proof.verify(
                    ledger_info_to_sync.ledger_info(),
                    Some(expected_start_version),
                ) {
                    Ok(()) => {
                        self.state_value_syncer
                            .set_transaction_output_to_sync(transaction_outputs_with_proof);
                    }
                    Err(error) => {
                        self.reset_active_stream(Some(NotificationAndFeedback::new(
                            notification_id,
                            NotificationFeedback::PayloadProofFailed,
                        )))
                        .await?;
                        return Err(Error::VerificationError(format!(
                            "Transaction outputs with proof is invalid! Error: {:?}",
                            error
                        )));
                    }
                }
            } else {
                self.reset_active_stream(Some(NotificationAndFeedback::new(
                    notification_id,
                    NotificationFeedback::InvalidPayloadData,
                )))
                .await?;
                return Err(Error::InvalidPayload(
                    "Payload does not contain a single transaction info!".into(),
                ));
            }
        } else {
            self.reset_active_stream(Some(NotificationAndFeedback::new(
                notification_id,
                NotificationFeedback::PayloadTypeIsIncorrect,
            )))
            .await?;
            return Err(Error::InvalidPayload(
                "Did not receive transaction outputs with proof!".into(),
            ));
        }

        Ok(())
    }

    /// Verifies the first payload version matches the version we wish to sync
    async fn verify_payload_start_version(
        &mut self,
        notification_id: NotificationId,
        payload_start_version: Option<Version>,
        expected_start_version: Version,
    ) -> Result<Version, Error> {
        if let Some(payload_start_version) = payload_start_version {
            if payload_start_version != expected_start_version {
                self.reset_active_stream(Some(NotificationAndFeedback::new(
                    notification_id,
                    NotificationFeedback::InvalidPayloadData,
                )))
                .await?;
                Err(Error::VerificationError(format!(
                    "The payload start version does not match the expected version! Start: {:?}, expected: {:?}",
                    payload_start_version, expected_start_version
                )))
            } else {
                Ok(payload_start_version)
            }
        } else {
            self.reset_active_stream(Some(NotificationAndFeedback::new(
                notification_id,
                NotificationFeedback::EmptyPayloadData,
            )))
            .await?;
            Err(Error::VerificationError(
                "The playload starting version is missing!".into(),
            ))
        }
    }

    /// Returns the end of epoch ledger info for the given payload. If
    /// calculation fails, the active stream is terminated. Assumes the
    /// `payload_start_version` exists.
    async fn get_end_of_epoch_ledger_info(
        &mut self,
        notification_id: NotificationId,
        payload_start_version: Version,
        transaction_list_with_proof: Option<&TransactionListWithProof>,
        transaction_outputs_with_proof: Option<&TransactionOutputListWithProof>,
    ) -> Result<Option<LedgerInfoWithSignatures>, Error> {
        // Calculate the payload end version
        let num_versions = match self.driver_configuration.config.bootstrapping_mode {
            BootstrappingMode::ApplyTransactionOutputsFromGenesis => {
                if let Some(transaction_outputs_with_proof) = transaction_outputs_with_proof {
                    transaction_outputs_with_proof
                        .transactions_and_outputs
                        .len()
                } else {
                    self.reset_active_stream(Some(NotificationAndFeedback::new(
                        notification_id,
                        NotificationFeedback::PayloadTypeIsIncorrect,
                    )))
                    .await?;
                    return Err(Error::InvalidPayload(
                        "Did not receive transaction outputs with proof!".into(),
                    ));
                }
            }
            BootstrappingMode::ExecuteTransactionsFromGenesis => {
                if let Some(transaction_list_with_proof) = transaction_list_with_proof {
                    transaction_list_with_proof.transactions.len()
                } else {
                    self.reset_active_stream(Some(NotificationAndFeedback::new(
                        notification_id,
                        NotificationFeedback::PayloadTypeIsIncorrect,
                    )))
                    .await?;
                    return Err(Error::InvalidPayload(
                        "Did not receive transactions with proof!".into(),
                    ));
                }
            }
            bootstrapping_mode => {
                unimplemented!("Bootstrapping mode not supported: {:?}", bootstrapping_mode)
            }
        };
        let payload_end_version = payload_start_version
            .checked_add(num_versions as u64)
            .and_then(|v| v.checked_sub(1))
            .ok_or_else(|| {
                Error::IntegerOverflow("The payload end version has overflown!".into())
            })?; // payload_end_version = payload_start_version + num_versions - 1

        // Find any epoch ending ledger info at the payload end version
        Ok(self
            .verified_epoch_states
            .get_epoch_ending_ledger_info(payload_end_version))
    }

    /// Returns the highest known ledger info (including the newly fetch ones)
    fn get_highest_known_ledger_info(&self) -> Result<LedgerInfoWithSignatures, Error> {
        // Fetch the highest synced ledger info from storage
        let mut highest_known_ledger_info =
            utils::fetch_latest_synced_ledger_info(self.storage.clone())?;

        // Fetch the highest verified ledger info (from the network) and take
        // the maximum.
        if let Some(verified_ledger_info) =
            self.verified_epoch_states.get_highest_known_ledger_info()
        {
            if verified_ledger_info.ledger_info().version()
                > highest_known_ledger_info.ledger_info().version()
            {
                highest_known_ledger_info = verified_ledger_info;
            }
        }
        Ok(highest_known_ledger_info)
    }

    /// Handles the end of stream notification or an invalid payload by
    /// terminating the stream appropriately.
    async fn handle_end_of_stream_or_invalid_payload(
        &mut self,
        data_notification: DataNotification,
    ) -> Result<(), Error> {
        // Calculate the feedback based on the notification
        let notification_feedback = match data_notification.data_payload {
            DataPayload::EndOfStream => NotificationFeedback::EndOfStream,
            _ => NotificationFeedback::PayloadTypeIsIncorrect,
        };
        let notification_and_feedback =
            NotificationAndFeedback::new(data_notification.notification_id, notification_feedback);

        // Reset the stream
        self.reset_active_stream(Some(notification_and_feedback))
            .await?;

        // Return an error if the payload was invalid
        match data_notification.data_payload {
            DataPayload::EndOfStream => Ok(()),
            _ => Err(Error::InvalidPayload("Unexpected payload type!".into())),
        }
    }

    /// Returns the speculative stream state. Assumes that the state exists.
    fn get_speculative_stream_state(&mut self) -> &mut SpeculativeStreamState {
        self.speculative_stream_state
            .as_mut()
            .expect("Speculative stream state does not exist!")
    }

    /// Resets the currently active data stream and speculative state
    pub async fn reset_active_stream(
        &mut self,
        notification_and_feedback: Option<NotificationAndFeedback>,
    ) -> Result<(), Error> {
        if let Some(active_data_stream) = &self.active_data_stream {
            let data_stream_id = active_data_stream.data_stream_id;
            utils::terminate_stream_with_feedback(
                &mut self.streaming_client,
                data_stream_id,
                notification_and_feedback,
            )
            .await?;
        }

        self.active_data_stream = None;
        self.speculative_stream_state = None;
        Ok(())
    }

    /// Returns the verified epoch states struct for testing purposes
    #[cfg(test)]
    pub(crate) fn get_verified_epoch_states(&mut self) -> &mut VerifiedEpochStates {
        &mut self.verified_epoch_states
    }

    /// Returns the state value syncer struct for testing purposes
    #[cfg(test)]
    pub(crate) fn get_state_value_syncer(&mut self) -> &mut StateValueSyncer {
        &mut self.state_value_syncer
    }
}


text:
/**
 * Canvas.java
 *
 * This file was auto-generated from WSDL
 * by the Apache Axis 1.4 Apr 22, 2006 (06:55:48 PDT) WSDL2Java emitter.
 */

package com.sforce.soap.partner;

public class Canvas  extends com.sforce.soap.partner.DescribeLayoutComponent  implements java.io.Serializable {
    private java.lang.String displayLocation;

    private java.lang.String referenceId;

    private boolean showLabel;

    private boolean showScrollbars;

    private java.lang.String suggestedHeight;

    private java.lang.String suggestedWidth;

    public Canvas() {
    }

    public Canvas(
           int displayLines,
           int tabOrder,
           com.sforce.soap.partner.LayoutComponentType type,
           java.lang.String value,
           java.lang.String displayLocation,
           java.lang.String referenceId,
           boolean showLabel,
           boolean showScrollbars,
           java.lang.String suggestedHeight,
           java.lang.String suggestedWidth) {
        super(
            displayLines,
            tabOrder,
            type,
            value);
        this.displayLocation = displayLocation;
        this.referenceId = referenceId;
        this.showLabel = showLabel;
        this.showScrollbars = showScrollbars;
        this.suggestedHeight = suggestedHeight;
        this.suggestedWidth = suggestedWidth;
    }

    /**
     * Gets the displayLocation value for this Canvas.
     *
     * @return displayLocation
     */
    public java.lang.String getDisplayLocation() {
        return displayLocation;
    }

    /**
     * Sets the displayLocation value for this Canvas.
     *
     * @param displayLocation
     */
    public void setDisplayLocation(java.lang.String displayLocation) {
        this.displayLocation = displayLocation;
    }

    /**
     * Gets the referenceId value for this Canvas.
     *
     * @return referenceId
     */
    public java.lang.String getReferenceId() {
        return referenceId;
    }

    /**
     * Sets the referenceId value for this Canvas.
     *
     * @param referenceId
     */
    public void setReferenceId(java.lang.String referenceId) {
        this.referenceId = referenceId;
    }

    /**
     * Gets the showLabel value for this Canvas.
     *
     * @return showLabel
     */
    public boolean isShowLabel() {
        return showLabel;
    }

    /**
     * Sets the showLabel value for this Canvas.
     *
     * @param showLabel
     */
    public void setShowLabel(boolean showLabel) {
        this.showLabel = showLabel;
    }

    /**
     * Gets the showScrollbars value for this Canvas.
     *
     * @return showScrollbars
     */
    public boolean isShowScrollbars() {
        return showScrollbars;
    }

    /**
     * Sets the showScrollbars value for this Canvas.
     *
     * @param showScrollbars
     */
    public void setShowScrollbars(boolean showScrollbars) {
        this.showScrollbars = showScrollbars;
    }

    /**
     * Gets the suggestedHeight value for this Canvas.
     *
     * @return suggestedHeight
     */
    public java.lang.String getSuggestedHeight() {
        return suggestedHeight;
    }

    /**
     * Sets the suggestedHeight value for this Canvas.
     *
     * @param suggestedHeight
     */
    public void setSuggestedHeight(java.lang.String suggestedHeight) {
        this.suggestedHeight = suggestedHeight;
    }

    /**
     * Gets the suggestedWidth value for this Canvas.
     *
     * @return suggestedWidth
     */
    public java.lang.String getSuggestedWidth() {
        return suggestedWidth;
    }

    /**
     * Sets the suggestedWidth value for this Canvas.
     *
     * @param suggestedWidth
     */
    public void setSuggestedWidth(java.lang.String suggestedWidth) {
        this.suggestedWidth = suggestedWidth;
    }

    private java.lang.Object __equalsCalc = null;
    public synchronized boolean equals(java.lang.Object obj) {
        if (!(obj instanceof Canvas)) return false;
        Canvas other = (Canvas) obj;
        if (obj == null) return false;
        if (this == obj) return true;
        if (__equalsCalc != null) {
            return (__equalsCalc == obj);
        }
        __equalsCalc = obj;
        boolean _equals;
        _equals = super.equals(obj) &&
            ((this.displayLocation==null && other.getDisplayLocation()==null) ||
             (this.displayLocation!=null &&
              this.displayLocation.equals(other.getDisplayLocation()))) &&
            ((this.referenceId==null && other.getReferenceId()==null) ||
             (this.referenceId!=null &&
              this.referenceId.equals(other.getReferenceId()))) &&
            this.showLabel == other.isShowLabel() &&
            this.showScrollbars == other.isShowScrollbars() &&
            ((this.suggestedHeight==null && other.getSuggestedHeight()==null) ||
             (this.suggestedHeight!=null &&
              this.suggestedHeight.equals(other.getSuggestedHeight()))) &&
            ((this.suggestedWidth==null && other.getSuggestedWidth()==null) ||
             (this.suggestedWidth!=null &&
              this.suggestedWidth.equals(other.getSuggestedWidth())));
        __equalsCalc = null;
        return _equals;
    }

    private boolean __hashCodeCalc = false;
    public synchronized int hashCode() {
        if (__hashCodeCalc) {
            return 0;
        }
        __hashCodeCalc = true;
        int _hashCode = super.hashCode();
        if (getDisplayLocation() != null) {
            _hashCode += getDisplayLocation().hashCode();
        }
        if (getReferenceId() != null) {
            _hashCode += getReferenceId().hashCode();
        }
        _hashCode += (isShowLabel() ? Boolean.TRUE : Boolean.FALSE).hashCode();
        _hashCode += (isShowScrollbars() ? Boolean.TRUE : Boolean.FALSE).hashCode();
        if (getSuggestedHeight() != null) {
            _hashCode += getSuggestedHeight().hashCode();
        }
        if (getSuggestedWidth() != null) {
            _hashCode += getSuggestedWidth().hashCode();
        }
        __hashCodeCalc = false;
        return _hashCode;
    }

    // Type metadata
    private static org.apache.axis.description.TypeDesc typeDesc =
        new org.apache.axis.description.TypeDesc(Canvas.class, true);

    static {
        typeDesc.setXmlType(new javax.xml.namespace.QName("urn:partner.soap.sforce.com", "Canvas"));
        org.apache.axis.description.ElementDesc elemField = new org.apache.axis.description.ElementDesc();
        elemField.setFieldName("displayLocation");
        elemField.setXmlName(new javax.xml.namespace.QName("urn:partner.soap.sforce.com", "displayLocation"));
        elemField.setXmlType(new javax.xml.namespace.QName("http://www.w3.org/2001/XMLSchema", "string"));
        elemField.setNillable(false);
        typeDesc.addFieldDesc(elemField);
        elemField = new org.apache.axis.description.ElementDesc();
        elemField.setFieldName("referenceId");
        elemField.setXmlName(new javax.xml.namespace.QName("urn:partner.soap.sforce.com", "referenceId"));
        elemField.setXmlType(new javax.xml.namespace.QName("http://www.w3.org/2001/XMLSchema", "string"));
        elemField.setNillable(false);
        typeDesc.addFieldDesc(elemField);
        elemField = new org.apache.axis.description.ElementDesc();
        elemField.setFieldName("showLabel");
        elemField.setXmlName(new javax.xml.namespace.QName("urn:partner.soap.sforce.com", "showLabel"));
        elemField.setXmlType(new javax.xml.namespace.QName("http://www.w3.org/2001/XMLSchema", "boolean"));
        elemField.setNillable(false);
        typeDesc.addFieldDesc(elemField);
        elemField = new org.apache.axis.description.ElementDesc();
        elemField.setFieldName("showScrollbars");
        elemField.setXmlName(new javax.xml.namespace.QName("urn:partner.soap.sforce.com", "showScrollbars"));
        elemField.setXmlType(new javax.xml.namespace.QName("http://www.w3.org/2001/XMLSchema", "boolean"));
        elemField.setNillable(false);
        typeDesc.addFieldDesc(elemField);
        elemField = new org.apache.axis.description.ElementDesc();
        elemField.setFieldName("suggestedHeight");
        elemField.setXmlName(new javax.xml.namespace.QName("urn:partner.soap.sforce.com", "suggestedHeight"));
        elemField.setXmlType(new javax.xml.namespace.QName("http://www.w3.org/2001/XMLSchema", "string"));
        elemField.setNillable(false);
        typeDesc.addFieldDesc(elemField);
        elemField = new org.apache.axis.description.ElementDesc();
        elemField.setFieldName("suggestedWidth");
        elemField.setXmlName(new javax.xml.namespace.QName("urn:partner.soap.sforce.com", "suggestedWidth"));
        elemField.setXmlType(new javax.xml.namespace.QName("http://www.w3.org/2001/XMLSchema", "string"));
        elemField.setNillable(false);
        typeDesc.addFieldDesc(elemField);
    }

    /**
     * Return type metadata object
     */
    public static org.apache.axis.description.TypeDesc getTypeDesc() {
        return typeDesc;
    }

    /**
     * Get Custom Serializer
     */
    public static org.apache.axis.encoding.Serializer getSerializer(
           java.lang.String mechType,
           java.lang.Class _javaType,
           javax.xml.namespace.QName _xmlType) {
        return
          new  org.apache.axis.encoding.ser.BeanSerializer(
            _javaType, _xmlType, typeDesc);
    }

    /**
     * Get Custom Deserializer
     */
    public static org.apache.axis.encoding.Deserializer getDeserializer(
           java.lang.String mechType,
           java.lang.Class _javaType,
           javax.xml.namespace.QName _xmlType) {
        return
          new  org.apache.axis.encoding.ser.BeanDeserializer(
            _javaType, _xmlType, typeDesc);
    }

}

redpajama_set_name: RedPajamaGithub


text:
class MemoryCache {
  static isExpired(entry) {
    return entry.expire && entry.expire < Date.now();
  }

  constructor(defaultTTL) {
    this.store = {};
    this.defaultTTL = defaultTTL;
  }

  set(key, value, ttl) {
    const finalTTL = ttl || this.defaultTTL;

    const entry = {
      value,
      expire: finalTTL ? finalTTL + Date.now() : false,
    };

    if (finalTTL) {
      entry.timeoutId = setTimeout(() => {
        this.remove(key);
      }, finalTTL);
    }

    this.store[key] = entry;

    return this;
  }

  has(key) {
    return (
      Object.keys(this.store).includes(key) &&
      !MemoryCache.isExpired(this.store[key])
    );
  }

  get(key, defaultValue) {
    const entry = this.store[key];

    if (!entry) {
      return defaultValue;
    }

    if (MemoryCache.isExpired(entry)) {
      this.remove(key);
      return defaultValue;
    }

    return entry.value;
  }

  remove(key) {
    if (!this.store[key]) {
      return this;
    }

    if (this.store[key].timeoutId) {
      clearTimeout(this.store[key].timeoutId);
    }

    delete this.store[key];

    return this;
  }

  clear() {
    this.store = {};

    return this;
  }
}

export default MemoryCache;


text:
package org.apache.jackrabbit.oak.security.authorization.evaluation;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;

import java.util.Set;

import com.google.common.collect.Sets;
import org.apache.jackrabbit.JcrConstants;
import org.apache.jackrabbit.oak.api.PropertyState;
import org.apache.jackrabbit.oak.api.Tree;
import org.apache.jackrabbit.oak.plugins.tree.TreeConstants;
import org.junit.Before;
import org.junit.Test;

/**
 * Test for the hidden {@link org.apache.jackrabbit.oak.plugins.tree.TreeConstants#OAK_CHILD_ORDER} property
 */
public class ChildOrderPropertyTest extends AbstractOakCoreTest {

    @Override
    @Before
    public void before() throws Exception {
        super.before();

        Tree a = root.getTree("/a");
        a.setOrderableChildren(true);
        root.commit();
    }

    @Test
    public void testHasProperty() {
        Tree a = root.getTree("/a");
        assertFalse(a.hasProperty(TreeConstants.OAK_CHILD_ORDER));
    }

    @Test
    public void testGetProperty() {
        Tree a = root.getTree("/a");
        assertNull(a.getProperty(TreeConstants.OAK_CHILD_ORDER));
    }

    @Test
    public void testGetProperties() {
        Set<String> propertyNames = Sets.newHashSet(JcrConstants.JCR_PRIMARYTYPE, "aProp");

        Tree a = root.getTree("/a");
        for (PropertyState prop : a.getProperties()) {
            assertTrue(propertyNames.remove(prop.getName()));
        }
        assertTrue(propertyNames.isEmpty());
    }

    @Test
    public void testGetPropertyCount() {
        Tree a = root.getTree("/a");
        assertEquals(2, a.getPropertyCount());
    }

    @Test
    public void testGetPropertyStatus() {
        Tree a = root.getTree("/a");
        assertNull(a.getPropertyStatus(TreeConstants.OAK_CHILD_ORDER));
    }

}


text:
.. This file is automatically generated. Do not edit this file directly.

Google Cloud Bigtable Python Samples
===============================================================================

This directory contains samples for Google Cloud Bigtable. `Google Cloud Bigtable`_ is Google's NoSQL Big Data database service. It's the same database that powers many core Google services, including Search, Analytics, Maps, and Gmail.

This sample demonstrates how to use `Stackdriver Monitoring`_
to scale Cloud Bigtable based on CPU usage.

.. _Stackdriver Monitoring: http://cloud.google.com/monitoring/docs/

.. _Google Cloud Bigtable: https://cloud.google.com/bigtable/docs/

Setup
-------------------------------------------------------------------------------

Authentication
++++++++++++++

Authentication is typically done through `Application Default Credentials`_,
which means you do not have to change the code to authenticate as long as
your environment has credentials. You have a few options for setting up
authentication:

#. When running locally, use the `Google Cloud SDK`_

    .. code-block:: bash

        gcloud auth application-default login

#. When running on App Engine or Compute Engine, credentials are already
   set-up. However, you may need to configure your Compute Engine instance
   with `additional scopes`_.

#. You can create a `Service Account key file`_. This file can be used to
   authenticate to Google Cloud Platform services from any environment. To use
   the file, set the ``GOOGLE_APPLICATION_CREDENTIALS`` environment variable to
   the path to the key file, for example:

    .. code-block:: bash

        export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service_account.json

.. _Application Default Credentials: https://cloud.google.com/docs/authentication#getting_credentials_for_server-centric_flow
.. _additional scopes: https://cloud.google.com/compute/docs/authentication#using
.. _Service Account key file: https://developers.google.com/identity/protocols/OAuth2ServiceAccount#creatinganaccount

Install Dependencies
++++++++++++++++++++

#. Install `pip`_ and `virtualenv`_ if you do not already have them.

#. Create a virtualenv. Samples are compatible with Python 2.7 and 3.4+.

    .. code-block:: bash

        $ virtualenv env
        $ source env/bin/activate

#. Install the dependencies needed to run the samples.

    .. code-block:: bash

        $ pip install -r requirements.txt

.. _pip: https://pip.pypa.io/
.. _virtualenv: https://virtualenv.pypa.io/

Samples
-------------------------------------------------------------------------------

Metricscaling example
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

To run this sample:

.. code-block:: bash

    $ python metricscaler.py

    usage: metricscaler.py [-h] [--high_cpu_threshold HIGH_CPU_THRESHOLD]
                           [--low_cpu_threshold LOW_CPU_THRESHOLD]
                           [--short_sleep SHORT_SLEEP] [--long_sleep LONG_SLEEP]
                           bigtable_instance bigtable_cluster

    Scales Cloud Bigtable clusters based on CPU usage.

    positional arguments:
      bigtable_instance     ID of the Cloud Bigtable instance to connect to.
      bigtable_cluster      ID of the Cloud Bigtable cluster to connect to.

    optional arguments:
      -h, --help            show this help message and exit
      --high_cpu_threshold HIGH_CPU_THRESHOLD
                            If Cloud Bigtable CPU usage is above this threshold,
                            scale up
      --low_cpu_threshold LOW_CPU_THRESHOLD
                            If Cloud Bigtable CPU usage is below this threshold,
                            scale down
      --short_sleep SHORT_SLEEP
                            How long to sleep in seconds between checking metrics
                            after no scale operation
      --long_sleep LONG_SLEEP
                            How long to sleep in seconds between checking metrics
                            after a scaling operation

The client library
-------------------------------------------------------------------------------

This sample uses the `Google Cloud Client Library for Python`_.
You can read the documentation for more details on API usage and use GitHub
to `browse the source`_ and  `report issues`_.

.. _Google Cloud Client Library for Python:
    https://googlecloudplatform.github.io/google-cloud-python/
.. _browse the source:
    https://github.com/GoogleCloudPlatform/google-cloud-python
.. _report issues:
    https://github.com/GoogleCloudPlatform/google-cloud-python/issues

.. _Google Cloud SDK: https://cloud.google.com/sdk/


text:
'use strict';

Object.defineProperty(exports, '__esModule', {
  value: true
});
exports.default = Timer;

// This file is a heavily modified fork of Jasmine. Original license:
/*
Copyright (c) 2008-2016 Pivotal Labs

Permission is hereby granted, free of charge, to any person obtaining
a copy of this software and associated documentation files (the
"Software"), to deal in the Software without restriction, including
without limitation the rights to use, copy, modify, merge, publish,
distribute, sublicense, and/or sell copies of the Software, and to
permit persons to whom the Software is furnished to do so, subject to
the following conditions:

The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/

/* eslint-disable sort-keys */

const defaultNow = (function(Date) {
  return function() {
    return new Date().getTime();
  };
})(Date);

function Timer(options) {
  options = options || {};

  const now = options.now || defaultNow;
  let startTime;

  this.start = function() {
    startTime = now();
  };

  this.elapsed = function() {
    return now() - startTime;
  };
}


text:
require "spec_helper"

if mongo_orm == "mongoid"
  describe MongoidStore do
    describe ".session_class" do
      subject { described_class.session_class }

      it "returns mongoid session class" do
        expect(subject).to eq(described_class::Session)
      end
    end
  end
end


text:
lesser
======

Lesser is a custom Wordpress theme for our blog.


text:
;
(function ($, window, document) {
    "use strict"; // jshint ;_;
    var PullToRefresh = function(options){
        var me = this;
        this.heightRange = 40;
        this.animationDuration = 80;
        this.done = false;
        this.startPos = {x: 0 , y: 0};
        this.time = 0;
        this.pulledClassName = "pull-to-refresh-status-pulled";
        this.loadingClassName = "pull-to-refresh-status-loading";
        if(options){
            $.extend(this, options);
        }
        $(document).on("touchstart", function(e){
            me.onTouchstart.apply(me, [e]);
        })
        .on("touchmove", function(e){
            me.onTouchmove.apply(me, [e]);
        })
        .on("touchend", function(e){
            me.onTouchend.apply(me, [e]);
        });
    }

    $.extend(PullToRefresh.prototype, {
        onTouchstart: function(e){
            if (e && e.touches){
                var touch = e.touches[0];
            }else{
                return;
            }
            this.startPos = {
                'x': touch.pageX,
                'y': touch.pageY
            }
            this.time = new Date().getTime();
            this.done = false;
        },
        onTouchmove: function(e){
            if (e && e.touches && e.touches.length > 0){
                var now = new Date().getTime();
                if ((now - this.time) < this.animationDuration){
                    return;
                }else{
                    this.time = now;
                }
                var touch = e.touches[0];
            }else{
                return;
            }
            this.pos = {
                'x': touch.pageX,
                'y': touch.pageY
            }
            var scrollY = this.pos.y - this.startPos.y
            if ( scrollY < 6) {
                return;
            }

            var top_ = parseInt(this.bind.css('top')) || 0;
            if( top_ < this.heightRange){
                this.bind.css('top', scrollY > this.heightRange ?
                        this.heightRange : scrollY);
                return;
            }else{
                this.done = true;
                $(".pull-to-refresh").addClass(this.pulledClassName)// change icon style
                return;
            }
        },
        onTouchend: function(e){
            if(this.done && this.callback){
                this.loading();
                this.callback.call(this);
            } else {
                this.cancel();
            }
        },
        loading: function(){
            $(".pull-to-refresh").removeClass(this.pulledClassName)// change icon style
                .addClass(this.loadingClassName)
        },
        back: function(){
            this.bind.css('top', 0);
            $(".pull-to-refresh").removeClass(this.pulledClassName)// change icon style
                .removeClass(this.loadingClassName)
        },
        cancel: function(){
            this.back();
        }
    });

    $.fn.pullToRefresh = function(options){
        var callback;
        if (typeof options == "function"){
            callback = options;
        }
        var bind = this.parent('.pullable');
        options = {
            "callback": callback,
            "target": this,
            "bind": bind
        }

        var pullToRefresh = new PullToRefresh(options);
        return pullToRefresh;
    }

}($, window, document));


text:
{% extends "layout-atom.xml" %}
{% from "helper.xml" import atom_item %}
{% block feed_title %}Last /sug/ threads{% endblock %}
{% block feed_description %}The latest /sug/ threads found on /co/ and /trash/{% endblock %}
{% block feed_update %}{{ lastupdate }}{% endblock %}
{% block feed_id %}https://sug.rocks/threads.xml{% endblock %}
{% block feed_self %}https://sug.rocks/threads.xml{% endblock %}
{% block feed_link %}https://sug.rocks/{% endblock %}
{% block feed_items %}
{% for threadid in threadlist %}
{% set thread = threaddict[threadid] %}
{% set published %}{{ thread['dates']['ISO8601'] }}{% endset %}
{% set updated %}{{ thread['dates']['ISO8601'] }}{% endset %}
{% set title %}[/{{ thread['board'] }}/] {{ thread['edition'] }} (No.{{ thread['id'] }}){% endset %}
{% set content %}{% if thread['media']['deleted'] == True %}

{% elif thread['media']['spoiler'] == True %}
                [Image was spoilered]
{% else %}

{% endif %}

                {{ thread['op']|inlinecss }}{% endset %}
{% set guid %}{{ thread['url'] }}{% endset %}
{% if thread['status']['closed'] == False %}
{{ atom_item(title, content, guid, published, updated) }}
{% endif %}
{% endfor %}
{% endblock %}


text:
Simple application that guesses gender based on height and weight

### Website
[https://guarded-cliffs-5378.herokuapp.com/](https://guarded-cliffs-5378.herokuapp.com/)


text:
module Nimbos::Concerns::GeneratePost

	extend ActiveSupport::Concern

	included do
		after_create  :generate_post
	end

  def generate_post
    Nimbos::Post.log(user_id, self, self.to_s, "created", true)
  end

end


text:
package org.estatio.integtests.document;

import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;

import java.util.List;

import javax.inject.Inject;

import org.junit.Before;
import org.junit.Test;

import org.apache.isis.applib.fixturescripts.FixtureScript;

import org.estatio.dom.document.Document;
import org.estatio.dom.document.Documents;
import org.estatio.fixture.EstatioBaseLineFixture;
import org.estatio.fixture.interactivemap.InteractiveMapDocumentForOxf;
import org.estatio.integtests.EstatioIntegrationTest;

public class DocumentsTest extends EstatioIntegrationTest {

    @Inject
    Documents documents;

    @Before
    public void setupData() {
        runScript(new FixtureScript() {
            @Override
            protected void execute(ExecutionContext executionContext) {
                executionContext.executeChild(this, new EstatioBaseLineFixture());
                executionContext.executeChild(this, new InteractiveMapDocumentForOxf());
            }
        });

    }

    public static class AllDocuments extends DocumentsTest {

        @Test
        public void allDocuments() throws Exception {
            // given
            // when
            final List<Document> result = documents.allDocuments();
            // then
            assertThat(result.size(), is(1));
        }
    }

    public static class FindByName extends DocumentsTest {

        @Test
        public void happyCase() throws Exception {
            // given
            // when
            final List<Document> result = documents.findByName("*V*");
            // then
            assertThat(result.size(), is(1));
        }
    }

}


text:
[![Crates.io](https://img.shields.io/crates/v/cellsplit.svg)](https://crates.io/crates/cellsplit)
[![Travis](https://img.shields.io/travis/durka/cellsplit.svg)](https://travis-ci.org/durka/cellsplit)

What?
=====

This program can split apart, and recombine, MATLAB scripts written using the Cell Mode feature.

Why?
====

Cell Mode is highly convenient for interactive development. A large task can be split into separate chunks which are repeatedly run independently, inspecting local variables and making iterative changes. I often use Cell Mode to document the steps of a machine learning pipeline. But Cell Mode can't be used when operating MATLAB from the comand line (such as on a remote server). You can only run the whole script as a unit. But after processing with _cellsplit_, each cell _is_ a whole script so the flexibility is regained.

Also, if you've ever tried to set `dbstop if error` and then run a cell, and watched your compiler slowly melt into a pile of jello, _cellsplit_ can help. Since the "cells" are run as normal scripts, debugging and profiling work fine.

How?
====

_cellsplit_ contains a very rudimentary parser of MATLAB syntax, and it breaks out every cell into a new script. It also breaks out bodies of conditionals and loops, since cells can be placed inside those.

Status
======

Unsupported features of MATLAB:

- `switch`
- (please file a bug if there is one I've overlooked)


text:
namespace chromecast {
namespace media {

namespace {

bool RectFEqual(const RectF& r1, const RectF& r2) {
  return r1.x == r2.x && r1.y == r2.y && r1.width == r2.width &&
         r1.height == r2.height;
}

bool SizeEqual(const Size& s1, const Size& s2) {
  return s1.width == s2.width && s1.height == s2.height;
}

bool DisplayRectFValid(const RectF& r) {
  return r.width >= 0 && r.height >= 0;
}

bool ResolutionSizeValid(const Size& s) {
  return s.width >= 0 && s.height >= 0;
}

// Translates a gfx::OverlayTransform into a VideoPlane::Transform.
// Could be just a lookup table once we have unit tests for this code
// to ensure it stays in sync with OverlayTransform.
chromecast::media::VideoPlane::Transform ConvertTransform(
    gfx::OverlayTransform transform) {
  switch (transform) {
    case gfx::OVERLAY_TRANSFORM_NONE:
      return chromecast::media::VideoPlane::TRANSFORM_NONE;
    case gfx::OVERLAY_TRANSFORM_FLIP_HORIZONTAL:
      return chromecast::media::VideoPlane::FLIP_HORIZONTAL;
    case gfx::OVERLAY_TRANSFORM_FLIP_VERTICAL:
      return chromecast::media::VideoPlane::FLIP_VERTICAL;
    case gfx::OVERLAY_TRANSFORM_ROTATE_90:
      return chromecast::media::VideoPlane::ROTATE_90;
    case gfx::OVERLAY_TRANSFORM_ROTATE_180:
      return chromecast::media::VideoPlane::ROTATE_180;
    case gfx::OVERLAY_TRANSFORM_ROTATE_270:
      return chromecast::media::VideoPlane::ROTATE_270;
    default:
      NOTREACHED();
      return chromecast::media::VideoPlane::TRANSFORM_NONE;
  }
}

}  // namespace

// Helper class for calling VideoPlane::SetGeometry with rate-limiting.
// SetGeometry can take on the order of 100ms to run in some implementations
// and can be called on the order of 20x / second (as fast as graphics frames
// are produced).  This creates an ever-growing backlog of tasks on the media
// thread.
// This class measures the time taken to run SetGeometry to determine a
// reasonable frequency at which to call it.  Excess calls are coalesced
// to just set the most recent geometry.
class VideoPlaneController::RateLimitedSetVideoPlaneGeometry
    : public base::RefCountedThreadSafe<RateLimitedSetVideoPlaneGeometry> {
 public:
  RateLimitedSetVideoPlaneGeometry(
      const scoped_refptr<base::SingleThreadTaskRunner>& task_runner)
      : pending_display_rect_(0, 0, 0, 0),
        pending_set_geometry_(false),
        min_calling_interval_ms_(0),
        sample_counter_(0),
        task_runner_(task_runner) {}

  void SetGeometry(const chromecast::RectF& display_rect,
                   VideoPlane::Transform transform) {
    DCHECK(task_runner_->BelongsToCurrentThread());
    DCHECK(DisplayRectFValid(display_rect));

    base::TimeTicks now = base::TimeTicks::Now();
    base::TimeDelta elapsed = now - last_set_geometry_time_;

    if (elapsed < base::TimeDelta::FromMilliseconds(min_calling_interval_ms_)) {
      if (!pending_set_geometry_) {
        pending_set_geometry_ = true;

        task_runner_->PostDelayedTask(
            FROM_HERE,
            base::BindOnce(
                &RateLimitedSetVideoPlaneGeometry::ApplyPendingSetGeometry,
                this),
            base::TimeDelta::FromMilliseconds(2 * min_calling_interval_ms_));
      }

      pending_display_rect_ = display_rect;
      pending_transform_ = transform;
      return;
    }
    last_set_geometry_time_ = now;

    LOG(INFO) << __FUNCTION__ << " rect=" << display_rect.width << "x"
              << display_rect.height << " @" << display_rect.x << ","
              << display_rect.y << " transform " << transform;

    VideoPlane* video_plane = CastMediaShlib::GetVideoPlane();
    CHECK(video_plane);
    base::TimeTicks start = base::TimeTicks::Now();
    video_plane->SetGeometry(display_rect, transform);

    base::TimeDelta set_geometry_time = base::TimeTicks::Now() - start;
    UpdateAverageTime(set_geometry_time.InMilliseconds());
  }

 private:
  friend class base::RefCountedThreadSafe<RateLimitedSetVideoPlaneGeometry>;
  ~RateLimitedSetVideoPlaneGeometry() {}

  void UpdateAverageTime(int64_t sample) {
    const size_t kSampleCount = 5;
    if (samples_.size() < kSampleCount)
      samples_.push_back(sample);
    else
      samples_[sample_counter_++ % kSampleCount] = sample;
    int64_t total = 0;
    for (int64_t s : samples_)
      total += s;
    min_calling_interval_ms_ = 2 * total / samples_.size();
  }

  void ApplyPendingSetGeometry() {
    if (pending_set_geometry_) {
      pending_set_geometry_ = false;
      SetGeometry(pending_display_rect_, pending_transform_);
    }
  }

  RectF pending_display_rect_;
  VideoPlane::Transform pending_transform_;
  bool pending_set_geometry_;
  base::TimeTicks last_set_geometry_time_;

  // Don't call SetGeometry faster than this interval.
  int64_t min_calling_interval_ms_;

  // Min calling interval is computed as double average of last few time samples
  // (i.e. allow at least as much time between calls as the call itself takes).
  std::vector<int64_t> samples_;
  size_t sample_counter_;

  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;

  DISALLOW_COPY_AND_ASSIGN(RateLimitedSetVideoPlaneGeometry);
};

VideoPlaneController::VideoPlaneController(
    const Size& graphics_resolution,
    scoped_refptr<base::SingleThreadTaskRunner> media_task_runner)
    : is_paused_(false),
      have_screen_res_(false),
      screen_res_(0, 0),
      graphics_plane_res_(graphics_resolution),
      have_video_plane_geometry_(false),
      video_plane_display_rect_(0, 0),
      video_plane_transform_(VideoPlane::TRANSFORM_NONE),
      media_task_runner_(media_task_runner),
      video_plane_wrapper_(
          new RateLimitedSetVideoPlaneGeometry(media_task_runner_)) {}

VideoPlaneController::~VideoPlaneController() {}

void VideoPlaneController::SetGeometry(const gfx::RectF& gfx_display_rect,
                                       gfx::OverlayTransform gfx_transform) {
  const RectF display_rect(gfx_display_rect.x(), gfx_display_rect.y(),
                           gfx_display_rect.width(), gfx_display_rect.height());
  VideoPlane::Transform transform = ConvertTransform(gfx_transform);

  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK(DisplayRectFValid(display_rect));
  if (have_video_plane_geometry_ &&
      RectFEqual(display_rect, video_plane_display_rect_) &&
      transform == video_plane_transform_) {
    DVLOG(2) << "No change found in geometry parameters.";
    return;
  }

  LOG(INFO) << "New geometry parameters "
            << " rect=" << display_rect.width << "x" << display_rect.height
            << " @" << display_rect.x << "," << display_rect.y << " transform "
            << transform;

  have_video_plane_geometry_ = true;
  video_plane_display_rect_ = display_rect;
  video_plane_transform_ = transform;

  MaybeRunSetGeometry();
}

void VideoPlaneController::SetScreenResolution(const Size& resolution) {
  DCHECK(thread_checker_.CalledOnValidThread());
  DCHECK(ResolutionSizeValid(resolution));
  if (have_screen_res_ && SizeEqual(resolution, screen_res_)) {
    DVLOG(2) << "No change found in screen resolution.";
    return;
  }

  LOG(INFO) << "New screen resolution " << resolution.width << "x"
            << resolution.height;

  have_screen_res_ = true;
  screen_res_ = resolution;

  MaybeRunSetGeometry();
}

void VideoPlaneController::Pause() {
  DCHECK(thread_checker_.CalledOnValidThread());
  LOG(INFO) << "Pausing controller. No more VideoPlane SetGeometry calls.";
  is_paused_ = true;
}

void VideoPlaneController::Resume() {
  DCHECK(thread_checker_.CalledOnValidThread());
  LOG(INFO) << "Resuming controller. VideoPlane SetGeometry calls are active.";
  is_paused_ = false;
  ClearVideoPlaneGeometry();
}

bool VideoPlaneController::is_paused() const {
  DCHECK(thread_checker_.CalledOnValidThread());
  return is_paused_;
}

void VideoPlaneController::MaybeRunSetGeometry() {
  if (is_paused_) {
    DVLOG(2)
        << "All VideoPlane SetGeometry calls are paused. Ignoring request.";
    return;
  }

  if (!HaveDataForSetGeometry()) {
    DVLOG(2) << "Don't have all VideoPlane SetGeometry data. Ignoring request.";
    return;
  }

  DCHECK(graphics_plane_res_.width != 0 && graphics_plane_res_.height != 0);

  RectF scaled_rect = video_plane_display_rect_;
  if (graphics_plane_res_.width != screen_res_.width ||
      graphics_plane_res_.height != screen_res_.height) {
    float sx =
        static_cast<float>(screen_res_.width) / graphics_plane_res_.width;
    float sy =
        static_cast<float>(screen_res_.height) / graphics_plane_res_.height;
    scaled_rect.x *= sx;
    scaled_rect.y *= sy;
    scaled_rect.width *= sx;
    scaled_rect.height *= sy;
  }

  media_task_runner_->PostTask(
      FROM_HERE, base::BindOnce(&RateLimitedSetVideoPlaneGeometry::SetGeometry,
                                video_plane_wrapper_, scaled_rect,
                                video_plane_transform_));
}

bool VideoPlaneController::HaveDataForSetGeometry() const {
  return have_screen_res_ && have_video_plane_geometry_;
}

void VideoPlaneController::ClearVideoPlaneGeometry() {
  have_video_plane_geometry_ = false;
  video_plane_display_rect_ = RectF(0, 0);
  video_plane_transform_ = VideoPlane::TRANSFORM_NONE;
}

}  // namespace media
}  // namespace chromecast