pub async fn write_lines(
        &self,
        db_name: &str,
        lines: &[ParsedLine<'_>],
        default_time: i64,
    ) -> Result<()> {
        // Return an error if this server is not yet ready
        let config = self.require_initialized()?;

        let db_name = DatabaseName::new(db_name).context(InvalidDatabaseName)?;
        let db = config
            .db_initialized(&db_name)
            .context(DatabaseNotFound { db_name: &*db_name })?;

        // need to split this in two blocks because we cannot hold a lock across an async call.
        let routing_config_target = {
            let rules = db.rules();
            if let Some(RoutingRules::RoutingConfig(routing_config)) = &rules.routing_rules {
                let sharded_entries = lines_to_sharded_entries(
                    lines,
                    default_time,
                    None as Option<&ShardConfig>,
                    &*rules,
                )
                .context(LineConversion)?;
                Some((routing_config.sink.clone(), sharded_entries))
            } else {
                None
            }
        };

        if let Some((sink, sharded_entries)) = routing_config_target {
            for i in sharded_entries {
                self.write_entry_sink(&db_name, &db, &sink, i.entry).await?;
            }
            return Ok(());
        }

        // Split lines into shards while holding a read lock on the sharding config.
        // Once the lock is released we have a vector of entries, each associated with a
        // shard id, and an Arc to the mapping between shard ids and node
        // groups. This map is atomically replaced every time the sharding
        // config is updated, hence it's safe to use after we release the shard config
        // lock.
        let (sharded_entries, shards) = {
            let rules = db.rules();

            let shard_config = rules.routing_rules.as_ref().map(|cfg| match cfg {
                RoutingRules::RoutingConfig(_) => unreachable!("routing config handled above"),
                RoutingRules::ShardConfig(shard_config) => shard_config,
            });

            let sharded_entries =
                lines_to_sharded_entries(lines, default_time, shard_config, &*rules)
                    .context(LineConversion)?;

            let shards = shard_config
                .as_ref()
                .map(|cfg| Arc::clone(&cfg.shards))
                .unwrap_or_default();

            (sharded_entries, shards)
        };

        // Write to all shards in parallel; as soon as one fails return error
        // immediately to the client and abort all other outstanding requests.
        // This can take some time, but we're no longer holding the lock to the shard
        // config.
        futures_util::future::try_join_all(
            sharded_entries
                .into_iter()
                .map(|e| self.write_sharded_entry(&db_name, &db, Arc::clone(&shards), e)),
        )
        .await?;

        Ok(())
    }
/// `write_lines` takes in raw line protocol and converts it to a collection
/// of ShardedEntry which are then sent to other IOx servers based on
/// the ShardConfig or sent to the local database for buffering in the
/// WriteBuffer and/or the MutableBuffer if configured.
///
/// The provided `default_time` is nanoseconds since the epoch and will be assigned
/// to any lines that don't have a timestamp.
`write_lines` takes in raw line protocol and converts it to a collection
of ShardedEntry which are then sent to other IOx servers based on
the ShardConfig or sent to the local database for buffering in the
WriteBuffer and/or the MutableBuffer if configured.

The provided `default_time` is nanoseconds since the epoch and will be assigned
to any lines that don't have a timestamp.
` write_lines ` takes in raw line protocol and converts it to a collection of ShardedEntry which are then sent to other IOx servers based on the ShardConfig or sent to the local database for buffering in the WriteBuffer and / or the MutableBuffer if configured . The provided ` default_time ` is nanoseconds since the epoch and will be assigned to any lines that don ' t have a timestamp .
pub async fn write_lines(
        &self,
        db_name: &str,
        lines: &[ParsedLine<'_>],
        default_time: i64,
    ) -> Result<()> {
        // Return an error if this server is not yet ready
        let config = self.require_initialized()?;
        let db_name = DatabaseName::new(db_name).context(InvalidDatabaseName)?;
        let db = config
            .db_initialized(&db_name)
            .context(DatabaseNotFound { db_name: &*db_name })?;
        // need to split this in two blocks because we cannot hold a lock across an async call.
        let routing_config_target = {
            let rules = db.rules();
            if let Some(RoutingRules::RoutingConfig(routing_config)) = &rules.routing_rules {
                let sharded_entries = lines_to_sharded_entries(
                    lines,
                    default_time,
                    None as Option<&ShardConfig>,
                    &*rules,
                )
                .context(LineConversion)?;
                Some((routing_config.sink.clone(), sharded_entries))
            } else {
                None
            }
        };
        if let Some((sink, sharded_entries)) = routing_config_target {
            for i in sharded_entries {
                self.write_entry_sink(&db_name, &db, &sink, i.entry).await?;
            }
            return Ok(());
        }
        // Split lines into shards while holding a read lock on the sharding config.
        // Once the lock is released we have a vector of entries, each associated with a
        // shard id, and an Arc to the mapping between shard ids and node
        // groups. This map is atomically replaced every time the sharding
        // config is updated, hence it's safe to use after we release the shard config
        // lock.
        let (sharded_entries, shards) = {
            let rules = db.rules();
            let shard_config = rules.routing_rules.as_ref().map(|cfg| match cfg {
                RoutingRules::RoutingConfig(_) => unreachable!("routing config handled above"),
                RoutingRules::ShardConfig(shard_config) => shard_config,
            });
            let sharded_entries =
                lines_to_sharded_entries(lines, default_time, shard_config, &*rules)
                    .context(LineConversion)?;
            let shards = shard_config
                .as_ref()
                .map(|cfg| Arc::clone(&cfg.shards))
                .unwrap_or_default();
            (sharded_entries, shards)
        };
        // Write to all shards in parallel; as soon as one fails return error
        // immediately to the client and abort all other outstanding requests.
        // This can take some time, but we're no longer holding the lock to the shard
        // config.
        futures_util::future::try_join_all(
            sharded_entries
                .into_iter()
                .map(|e| self.write_sharded_entry(&db_name, &db, Arc::clone(&shards), e)),
        )
        .await?;
        Ok(())
    }
pub async fn write_lines ( & self , db_name : & str , lines : & [ ParsedLine < ' _ > ] , default_time : i64 , ) -> Result < ( ) > { // Return an error if this server is not yet ready let config = self . require_initialized ( ) ? ; let db_name = DatabaseName :: new ( db_name ) . context ( InvalidDatabaseName ) ? ; let db = config . db_initialized ( & db_name ) . context ( DatabaseNotFound { db_name : & * db_name } ) ? ; // need to split this in two blocks because we cannot hold a lock across an async call. let routing_config_target = { let rules = db . rules ( ) ; if let Some ( RoutingRules :: RoutingConfig ( routing_config ) ) = & rules . routing_rules { let sharded_entries = lines_to_sharded_entries ( lines , default_time , None as Option < & ShardConfig > , & * rules , ) . context ( LineConversion ) ? ; Some ( ( routing_config . sink . clone ( ) , sharded_entries ) ) } else { None } } ; if let Some ( ( sink , sharded_entries ) ) = routing_config_target { for i in sharded_entries { self . write_entry_sink ( & db_name , & db , & sink , i . entry ) . await ? ; } return Ok ( ( ) ) ; } // Split lines into shards while holding a read lock on the sharding config. // Once the lock is released we have a vector of entries, each associated with a // shard id, and an Arc to the mapping between shard ids and node // groups. This map is atomically replaced every time the sharding // config is updated, hence it's safe to use after we release the shard config // lock. let ( sharded_entries , shards ) = { let rules = db . rules ( ) ; let shard_config = rules . routing_rules . as_ref ( ) . map ( | cfg | match cfg { RoutingRules :: RoutingConfig ( _ ) => unreachable ! ( " " ) , RoutingRules :: ShardConfig ( shard_config ) => shard_config , } ) ; let sharded_entries = lines_to_sharded_entries ( lines , default_time , shard_config , & * rules ) . context ( LineConversion ) ? ; let shards = shard_config . as_ref ( ) . map ( | cfg | Arc :: clone ( & cfg . shards ) ) . unwrap_or_default ( ) ; ( sharded_entries , shards ) } ; // Write to all shards in parallel; as soon as one fails return error // immediately to the client and abort all other outstanding requests. // This can take some time, but we're no longer holding the lock to the shard // config. futures_util :: future :: try_join_all ( sharded_entries . into_iter ( ) . map ( | e | self . write_sharded_entry ( & db_name , & db , Arc :: clone ( & shards ) , e ) ) , ) . await ? ; Ok ( ( ) ) }
`write_lines` takes in raw line protocol and converts it to a collection
of ShardedEntry which are then sent to other IOx servers based on
the ShardConfig or sent to the local database for buffering in the
WriteBuffer and/or the MutableBuffer if configured.
` write_lines ` takes in raw line protocol and converts it to a collection of ShardedEntry which are then sent to other IOx servers based on the ShardConfig or sent to the local database for buffering in the WriteBuffer and / or the MutableBuffer if configured .
&self
db_name
&str
lines
&[ParsedLine<'_>]
default_time
i64
&self
db_name
&str
lines
&[ParsedLine<'_>]
default_time
i64