pub async fn create<S>(
        topic: S,
        partition: Size,
        base_offset: Offset,
        option: &ConfigOption,
    ) -> Result<FileReplica, StorageError>
    where
        S: AsRef<str> + Send + 'static,
    {
        let replica_dir = option.base_dir.join(replica_dir_name(topic, partition));

        debug!("creating rep dir: {}", replica_dir.display());
        create_dir_all(&replica_dir).await?; // ensure dir_name exits

        let mut rep_option = option.clone();
        rep_option.base_dir = replica_dir;
        // create active segment

        let (segments, last_offset_res) = SegmentList::from_dir(&rep_option).await?;

        let active_segment = if let Some(last_offset) = last_offset_res {
            trace!("last segment found, validating offsets: {}", last_offset);
            let mut last_segment =
                MutableSegment::open_for_write(last_offset, &rep_option).await?;
            last_segment.validate().await?;
            trace!(
                "segment validated with last offset: {}",
                last_segment.get_end_offset()
            );
            last_segment
        } else {
            debug!("no segment found, creating new one");
            MutableSegment::create(base_offset, &rep_option).await?
        };

        let last_base_offset = active_segment.get_base_offset();

        let commit_checkpoint: CheckPoint<Offset> = CheckPoint::create(
            &rep_option,
            "replication.chk",
            last_base_offset
        ).await?;

        Ok(FileReplica {
            option: rep_option,
            last_base_offset,
            partition,
            active_segment,
            prev_segments: segments,
            commit_checkpoint,
        })
    }
/// Construct a new replica with specified topic and partition.
/// It can start with arbitrary offset.  However, for normal replica,
/// it is usually starts with 0.  
///
/// Replica is minimum unit of logs that will can be replicated.  
/// It is a unique pair of (topic,partition)
///
/// Replica will use base directory to create it's own directory.
/// Directory name will encode unique replica id which is combination of topic
/// and partition.
///
/// If there is existing directory then it will load existing logs.
/// The logs will be validated to ensure it's safe to use it.
/// It is possible logs can't be used because they may be corrupted.
Construct a new replica with specified topic and partition.
It can start with arbitrary offset.  However, for normal replica,
it is usually starts with 0.

Replica is minimum unit of logs that will can be replicated.
It is a unique pair of (topic,partition)

Replica will use base directory to create it's own directory.
Directory name will encode unique replica id which is combination of topic
and partition.

If there is existing directory then it will load existing logs.
The logs will be validated to ensure it's safe to use it.
It is possible logs can't be used because they may be corrupted.
Construct a new replica with specified topic and partition . It can start with arbitrary offset . However for normal replica it is usually starts with 0 . Replica is minimum unit of logs that will can be replicated . It is a unique pair of ( topic partition ) Replica will use base directory to create it ' s own directory . Directory name will encode unique replica id which is combination of topic and partition . If there is existing directory then it will load existing logs . The logs will be validated to ensure it ' s safe to use it . It is possible logs can ' t be used because they may be corrupted .
pub async fn create<S>(
        topic: S,
        partition: Size,
        base_offset: Offset,
        option: &ConfigOption,
    ) -> Result<FileReplica, StorageError>
    where
        S: AsRef<str> + Send + 'static,
    {
        let replica_dir = option.base_dir.join(replica_dir_name(topic, partition));
        debug!("creating rep dir: {}", replica_dir.display());
        create_dir_all(&replica_dir).await?; // ensure dir_name exits
        let mut rep_option = option.clone();
        rep_option.base_dir = replica_dir;
        // create active segment
        let (segments, last_offset_res) = SegmentList::from_dir(&rep_option).await?;
        let active_segment = if let Some(last_offset) = last_offset_res {
            trace!("last segment found, validating offsets: {}", last_offset);
            let mut last_segment =
                MutableSegment::open_for_write(last_offset, &rep_option).await?;
            last_segment.validate().await?;
            trace!(
                "segment validated with last offset: {}",
                last_segment.get_end_offset()
            );
            last_segment
        } else {
            debug!("no segment found, creating new one");
            MutableSegment::create(base_offset, &rep_option).await?
        };
        let last_base_offset = active_segment.get_base_offset();
        let commit_checkpoint: CheckPoint<Offset> = CheckPoint::create(
            &rep_option,
            "replication.chk",
            last_base_offset
        ).await?;
        Ok(FileReplica {
            option: rep_option,
            last_base_offset,
            partition,
            active_segment,
            prev_segments: segments,
            commit_checkpoint,
        })
    }
pub async fn create < S > ( topic : S , partition : Size , base_offset : Offset , option : & ConfigOption , ) -> Result < FileReplica , StorageError > where S : AsRef < str > + Send + ' static , { let replica_dir = option . base_dir . join ( replica_dir_name ( topic , partition ) ) ; debug ! ( " " replica_dir display ( ) ) ; create_dir_all ( & replica_dir ) . await ? ; // ensure dir_name exits let mut rep_option = option . clone ( ) ; rep_option . base_dir = replica_dir ; // create active segment let ( segments , last_offset_res ) = SegmentList :: from_dir ( & rep_option ) . await ? ; let active_segment = if let Some ( last_offset ) = last_offset_res { trace ! ( " " last_offset ) ; let mut last_segment = MutableSegment :: open_for_write ( last_offset , & rep_option ) . await ? ; last_segment . validate ( ) . await ? ; trace ! ( " " last_segment get_end_offset ( ) ) ; last_segment } else { debug ! ( " " ) ; MutableSegment :: create ( base_offset , & rep_option ) . await ? } ; let last_base_offset = active_segment . get_base_offset ( ) ; let commit_checkpoint : CheckPoint < Offset > = CheckPoint :: create ( & rep_option , " " , last_base_offset ) . await ? ; Ok ( FileReplica { option : rep_option , last_base_offset , partition , active_segment , prev_segments : segments , commit_checkpoint , } ) }
Construct a new replica with specified topic and partition.
Construct a new replica with specified topic and partition .
topic
S
partition
Size
base_offset
Offset
option
&ConfigOption
topic
S
partition
Size
base_offset
Offset
option
&ConfigOption