Struct async_nats::ServerAddr
source · pub struct ServerAddr(_);Expand description
Address of a NATS server.
Implementations§
source§impl ServerAddr
impl ServerAddr
sourcepub fn from_url(url: Url) -> Result<Self>
pub fn from_url(url: Url) -> Result<Self>
Check if the URL is a valid NATS server address.
Examples found in repository?
src/lib.rs (line 1054)
1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055
fn from_str(input: &str) -> Result<Self, Self::Err> {
let url: Url = if input.contains("://") {
input.parse()
} else {
format!("nats://{input}").parse()
}
.map_err(|e| {
io::Error::new(
ErrorKind::InvalidInput,
format!("NATS server URL is invalid: {e}"),
)
})?;
Self::from_url(url)
}sourcepub fn into_inner(self) -> Url
pub fn into_inner(self) -> Url
Turn the server address into a standard URL.
sourcepub fn tls_required(&self) -> bool
pub fn tls_required(&self) -> bool
Returns if tls is required by the client for this server.
Examples found in repository?
src/connector.rs (line 120)
99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243
pub(crate) async fn try_connect(&mut self) -> Result<(ServerInfo, Connection), io::Error> {
let mut error = None;
let server_addrs: Vec<ServerAddr> = self.servers.keys().cloned().collect();
for server_addr in server_addrs {
let server_attempts = self.servers.get_mut(&server_addr).unwrap();
let duration = if *server_attempts == 0 {
Duration::from_millis(0)
} else {
let exp: u32 = (*server_attempts - 1).try_into().unwrap_or(std::u32::MAX);
let max = Duration::from_secs(4);
cmp::min(Duration::from_millis(2_u64.saturating_pow(exp)), max)
};
*server_attempts += 1;
sleep(duration).await;
let socket_addrs = server_addr.socket_addrs()?;
for socket_addr in socket_addrs {
match self
.try_connect_to(&socket_addr, server_addr.tls_required(), server_addr.host())
.await
{
Ok((server_info, mut connection)) => {
for url in &server_info.connect_urls {
let server_addr = url.parse::<ServerAddr>()?;
self.servers.entry(server_addr).or_insert(0);
}
let server_attempts = self.servers.get_mut(&server_addr).unwrap();
*server_attempts = 0;
let tls_required = self.options.tls_required || server_addr.tls_required();
let mut connect_info = ConnectInfo {
tls_required,
// FIXME(tp): have optional name
name: self.options.name.clone(),
pedantic: false,
verbose: false,
lang: LANG.to_string(),
version: VERSION.to_string(),
protocol: Protocol::Dynamic,
user: None,
pass: None,
auth_token: None,
user_jwt: None,
nkey: None,
signature: None,
echo: !self.options.no_echo,
headers: true,
no_responders: true,
};
match &self.options.auth {
Authorization::None => {
connection.write_op(ClientOp::Connect(connect_info)).await?;
self.events_tx.send(Event::Connected).await.ok();
self.state_tx.send(State::Connected).ok();
return Ok((server_info, connection));
}
Authorization::Token(token) => {
connect_info.auth_token = Some(token.clone())
}
Authorization::UserAndPassword(user, pass) => {
connect_info.user = Some(user.clone());
connect_info.pass = Some(pass.clone());
}
Authorization::NKey(ref seed) => {
match nkeys::KeyPair::from_seed(seed.as_str()) {
Ok(key_pair) => {
let nonce = server_info.nonce.clone();
match key_pair.sign(nonce.as_bytes()) {
Ok(signed) => {
connect_info.nkey = Some(key_pair.public_key());
connect_info.signature =
Some(base64_url::encode(&signed));
}
Err(e) => {
return Err(std::io::Error::new(
ErrorKind::Other,
format!(
"NKey auth: failed signing the nonce: {e}"
),
));
}
};
}
Err(e) => {
return Err(std::io::Error::new(
ErrorKind::Other,
format!("NKey auth: failed signing the nonce: {e}"),
));
}
}
}
Authorization::Jwt(jwt, sign_fn) => {
match sign_fn.call(server_info.nonce.clone()).await {
Ok(sig) => {
connect_info.user_jwt = Some(jwt.clone());
connect_info.signature = Some(sig);
}
Err(e) => {
return Err(std::io::Error::new(
ErrorKind::Other,
format!("JWT auth: failed signing the nonce: {e}"),
));
}
}
}
}
connection.write_op(ClientOp::Connect(connect_info)).await?;
connection.write_op(ClientOp::Ping).await?;
connection.flush().await?;
match connection.read_op().await? {
Some(ServerOp::Error(err)) => {
return Err(io::Error::new(
ErrorKind::InvalidInput,
err.to_string(),
));
}
Some(_) => {
self.events_tx.send(Event::Connected).await.ok();
self.state_tx.send(State::Connected).ok();
return Ok((server_info, connection));
}
None => {
return Err(io::Error::new(
ErrorKind::BrokenPipe,
"connection aborted",
))
}
}
}
Err(inner) => error.replace(inner),
};
}
}
Err(error.unwrap())
}sourcepub fn has_user_pass(&self) -> bool
pub fn has_user_pass(&self) -> bool
Returns if the server url had embedded username and password.
sourcepub fn host(&self) -> &str
pub fn host(&self) -> &str
Returns the host.
Examples found in repository?
More examples
src/connector.rs (line 120)
99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243
pub(crate) async fn try_connect(&mut self) -> Result<(ServerInfo, Connection), io::Error> {
let mut error = None;
let server_addrs: Vec<ServerAddr> = self.servers.keys().cloned().collect();
for server_addr in server_addrs {
let server_attempts = self.servers.get_mut(&server_addr).unwrap();
let duration = if *server_attempts == 0 {
Duration::from_millis(0)
} else {
let exp: u32 = (*server_attempts - 1).try_into().unwrap_or(std::u32::MAX);
let max = Duration::from_secs(4);
cmp::min(Duration::from_millis(2_u64.saturating_pow(exp)), max)
};
*server_attempts += 1;
sleep(duration).await;
let socket_addrs = server_addr.socket_addrs()?;
for socket_addr in socket_addrs {
match self
.try_connect_to(&socket_addr, server_addr.tls_required(), server_addr.host())
.await
{
Ok((server_info, mut connection)) => {
for url in &server_info.connect_urls {
let server_addr = url.parse::<ServerAddr>()?;
self.servers.entry(server_addr).or_insert(0);
}
let server_attempts = self.servers.get_mut(&server_addr).unwrap();
*server_attempts = 0;
let tls_required = self.options.tls_required || server_addr.tls_required();
let mut connect_info = ConnectInfo {
tls_required,
// FIXME(tp): have optional name
name: self.options.name.clone(),
pedantic: false,
verbose: false,
lang: LANG.to_string(),
version: VERSION.to_string(),
protocol: Protocol::Dynamic,
user: None,
pass: None,
auth_token: None,
user_jwt: None,
nkey: None,
signature: None,
echo: !self.options.no_echo,
headers: true,
no_responders: true,
};
match &self.options.auth {
Authorization::None => {
connection.write_op(ClientOp::Connect(connect_info)).await?;
self.events_tx.send(Event::Connected).await.ok();
self.state_tx.send(State::Connected).ok();
return Ok((server_info, connection));
}
Authorization::Token(token) => {
connect_info.auth_token = Some(token.clone())
}
Authorization::UserAndPassword(user, pass) => {
connect_info.user = Some(user.clone());
connect_info.pass = Some(pass.clone());
}
Authorization::NKey(ref seed) => {
match nkeys::KeyPair::from_seed(seed.as_str()) {
Ok(key_pair) => {
let nonce = server_info.nonce.clone();
match key_pair.sign(nonce.as_bytes()) {
Ok(signed) => {
connect_info.nkey = Some(key_pair.public_key());
connect_info.signature =
Some(base64_url::encode(&signed));
}
Err(e) => {
return Err(std::io::Error::new(
ErrorKind::Other,
format!(
"NKey auth: failed signing the nonce: {e}"
),
));
}
};
}
Err(e) => {
return Err(std::io::Error::new(
ErrorKind::Other,
format!("NKey auth: failed signing the nonce: {e}"),
));
}
}
}
Authorization::Jwt(jwt, sign_fn) => {
match sign_fn.call(server_info.nonce.clone()).await {
Ok(sig) => {
connect_info.user_jwt = Some(jwt.clone());
connect_info.signature = Some(sig);
}
Err(e) => {
return Err(std::io::Error::new(
ErrorKind::Other,
format!("JWT auth: failed signing the nonce: {e}"),
));
}
}
}
}
connection.write_op(ClientOp::Connect(connect_info)).await?;
connection.write_op(ClientOp::Ping).await?;
connection.flush().await?;
match connection.read_op().await? {
Some(ServerOp::Error(err)) => {
return Err(io::Error::new(
ErrorKind::InvalidInput,
err.to_string(),
));
}
Some(_) => {
self.events_tx.send(Event::Connected).await.ok();
self.state_tx.send(State::Connected).ok();
return Ok((server_info, connection));
}
None => {
return Err(io::Error::new(
ErrorKind::BrokenPipe,
"connection aborted",
))
}
}
}
Err(inner) => error.replace(inner),
};
}
}
Err(error.unwrap())
}sourcepub fn socket_addrs(&self) -> Result<impl Iterator<Item = SocketAddr>>
pub fn socket_addrs(&self) -> Result<impl Iterator<Item = SocketAddr>>
Return the sockets from resolving the server address.
Examples found in repository?
src/connector.rs (line 117)
99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243
pub(crate) async fn try_connect(&mut self) -> Result<(ServerInfo, Connection), io::Error> {
let mut error = None;
let server_addrs: Vec<ServerAddr> = self.servers.keys().cloned().collect();
for server_addr in server_addrs {
let server_attempts = self.servers.get_mut(&server_addr).unwrap();
let duration = if *server_attempts == 0 {
Duration::from_millis(0)
} else {
let exp: u32 = (*server_attempts - 1).try_into().unwrap_or(std::u32::MAX);
let max = Duration::from_secs(4);
cmp::min(Duration::from_millis(2_u64.saturating_pow(exp)), max)
};
*server_attempts += 1;
sleep(duration).await;
let socket_addrs = server_addr.socket_addrs()?;
for socket_addr in socket_addrs {
match self
.try_connect_to(&socket_addr, server_addr.tls_required(), server_addr.host())
.await
{
Ok((server_info, mut connection)) => {
for url in &server_info.connect_urls {
let server_addr = url.parse::<ServerAddr>()?;
self.servers.entry(server_addr).or_insert(0);
}
let server_attempts = self.servers.get_mut(&server_addr).unwrap();
*server_attempts = 0;
let tls_required = self.options.tls_required || server_addr.tls_required();
let mut connect_info = ConnectInfo {
tls_required,
// FIXME(tp): have optional name
name: self.options.name.clone(),
pedantic: false,
verbose: false,
lang: LANG.to_string(),
version: VERSION.to_string(),
protocol: Protocol::Dynamic,
user: None,
pass: None,
auth_token: None,
user_jwt: None,
nkey: None,
signature: None,
echo: !self.options.no_echo,
headers: true,
no_responders: true,
};
match &self.options.auth {
Authorization::None => {
connection.write_op(ClientOp::Connect(connect_info)).await?;
self.events_tx.send(Event::Connected).await.ok();
self.state_tx.send(State::Connected).ok();
return Ok((server_info, connection));
}
Authorization::Token(token) => {
connect_info.auth_token = Some(token.clone())
}
Authorization::UserAndPassword(user, pass) => {
connect_info.user = Some(user.clone());
connect_info.pass = Some(pass.clone());
}
Authorization::NKey(ref seed) => {
match nkeys::KeyPair::from_seed(seed.as_str()) {
Ok(key_pair) => {
let nonce = server_info.nonce.clone();
match key_pair.sign(nonce.as_bytes()) {
Ok(signed) => {
connect_info.nkey = Some(key_pair.public_key());
connect_info.signature =
Some(base64_url::encode(&signed));
}
Err(e) => {
return Err(std::io::Error::new(
ErrorKind::Other,
format!(
"NKey auth: failed signing the nonce: {e}"
),
));
}
};
}
Err(e) => {
return Err(std::io::Error::new(
ErrorKind::Other,
format!("NKey auth: failed signing the nonce: {e}"),
));
}
}
}
Authorization::Jwt(jwt, sign_fn) => {
match sign_fn.call(server_info.nonce.clone()).await {
Ok(sig) => {
connect_info.user_jwt = Some(jwt.clone());
connect_info.signature = Some(sig);
}
Err(e) => {
return Err(std::io::Error::new(
ErrorKind::Other,
format!("JWT auth: failed signing the nonce: {e}"),
));
}
}
}
}
connection.write_op(ClientOp::Connect(connect_info)).await?;
connection.write_op(ClientOp::Ping).await?;
connection.flush().await?;
match connection.read_op().await? {
Some(ServerOp::Error(err)) => {
return Err(io::Error::new(
ErrorKind::InvalidInput,
err.to_string(),
));
}
Some(_) => {
self.events_tx.send(Event::Connected).await.ok();
self.state_tx.send(State::Connected).ok();
return Ok((server_info, connection));
}
None => {
return Err(io::Error::new(
ErrorKind::BrokenPipe,
"connection aborted",
))
}
}
}
Err(inner) => error.replace(inner),
};
}
}
Err(error.unwrap())
}Trait Implementations§
source§impl Clone for ServerAddr
impl Clone for ServerAddr
source§fn clone(&self) -> ServerAddr
fn clone(&self) -> ServerAddr
Returns a copy of the value. Read more
1.0.0 · source§fn clone_from(&mut self, source: &Self)
fn clone_from(&mut self, source: &Self)
Performs copy-assignment from
source. Read moresource§impl Debug for ServerAddr
impl Debug for ServerAddr
source§impl FromStr for ServerAddr
impl FromStr for ServerAddr
source§impl Hash for ServerAddr
impl Hash for ServerAddr
source§impl PartialEq<ServerAddr> for ServerAddr
impl PartialEq<ServerAddr> for ServerAddr
source§fn eq(&self, other: &ServerAddr) -> bool
fn eq(&self, other: &ServerAddr) -> bool
This method tests for
self and other values to be equal, and is used
by ==.source§impl ToServerAddrs for ServerAddr
impl ToServerAddrs for ServerAddr
§type Iter = IntoIter<ServerAddr>
type Iter = IntoIter<ServerAddr>
Returned iterator over socket addresses which this type may correspond
to.