alex/beta-minecraft-rust
Minecraft Beta 1.7.3 Rewrite
BRANCH main
SNAPSHOT Draw terrain GPU-driven from packed per-section buffers with GPU occlusion culling, add a render distance slider (2-128) with a quad/disk shape, cut chunk bookkeeping at high render distances, and fix a light underflow at y=0
DOWNLOAD
root crates client src app remote.rs
use std::io::{self, BufReader, BufWriter, Read, Write};
use std::net::{Shutdown, TcpStream, ToSocketAddrs};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use common::protocol::{Bridge, Connection, PROTOCOL_VERSION, Packet};
pub struct Remote {
cancelled: Arc<AtomicBool>,
socket: Arc<Mutex<Option<TcpStream>>>,
}
impl Remote {
pub fn connect(address: &str, username: &str) -> io::Result<(Self, Connection)> {
let (host, port) = parse_address(address)?;
let username = username.to_owned();
let (bridge, client) = common::protocol::bridged_connection();
let cancelled = Arc::new(AtomicBool::new(false));
let socket = Arc::new(Mutex::new(None));
std::thread::Builder::new()
.name("server-connection".into())
.spawn({
let cancelled = cancelled.clone();
let socket = socket.clone();
move || {
let result = connect(&host, port, &username, &bridge, &cancelled, &socket);
if !cancelled.load(Ordering::Relaxed)
&& let Err(error) = result
{
let _ = bridge.incoming.send(Packet::Disconnect {
reason: error.to_string(),
});
}
if let Some(stream) = socket.lock().unwrap().take() {
let _ = stream.shutdown(Shutdown::Both);
}
}
})?;
Ok((Self { cancelled, socket }, client))
}
pub fn connected(&self) -> bool {
self.socket.lock().unwrap().is_some()
}
pub fn disconnect(&mut self) {
self.cancelled.store(true, Ordering::Relaxed);
if let Some(stream) = self.socket.lock().unwrap().as_ref() {
let _ = stream.shutdown(Shutdown::Both);
}
}
}
impl Drop for Remote {
fn drop(&mut self) {
self.disconnect();
}
}
fn connect(
host: &str,
port: u16,
username: &str,
bridge: &Bridge,
cancelled: &Arc<AtomicBool>,
socket: &Mutex<Option<TcpStream>>,
) -> Result<(), Box<dyn std::error::Error>> {
let mut stream = Err(io::Error::new(
io::ErrorKind::AddrNotAvailable,
"Server address has no IP addresses",
));
for address in (host, port).to_socket_addrs()? {
if cancelled.load(Ordering::Relaxed) {
return Ok(());
}
stream = TcpStream::connect_timeout(&address, Duration::from_secs(5));
if stream.is_ok() {
break;
}
}
let stream = stream?;
stream.set_nodelay(true)?;
stream.set_read_timeout(Some(Duration::from_secs(30)))?;
stream.set_write_timeout(Some(Duration::from_secs(5)))?;
{
let mut socket = socket.lock().unwrap();
if cancelled.load(Ordering::Relaxed) {
return Ok(());
}
*socket = Some(stream.try_clone()?);
}
let mut output = BufWriter::new(stream.try_clone()?);
let mut input = BufReader::new(stream);
login(&mut input, &mut output, username)?;
let sender = bridge.incoming.clone();
let reader_cancelled = cancelled.clone();
let reader = std::thread::Builder::new()
.name("server-packets".into())
.spawn(move || {
while !reader_cancelled.load(Ordering::Relaxed) {
match Packet::read_from_server(&mut input) {
Ok(packet) => {
let disconnected = matches!(packet, Packet::Disconnect { .. });
if sender.send(packet).is_err() || disconnected {
break;
}
}
Err(error) => {
let _ = sender.send(Packet::Disconnect {
reason: error.to_string(),
});
break;
}
}
}
})?;
let result = (|| -> io::Result<()> {
while !cancelled.load(Ordering::Relaxed) && !reader.is_finished() {
match bridge.outgoing.recv_timeout(Duration::from_millis(50)) {
Ok(packet) => {
packet.write(&mut output)?;
output.flush()?;
}
Err(mpsc::RecvTimeoutError::Timeout) => {}
Err(mpsc::RecvTimeoutError::Disconnected) => break,
}
}
Ok(())
})();
let _ = output.get_ref().shutdown(Shutdown::Both);
let _ = reader.join();
result.map_err(Into::into)
}
fn login(
input: &mut impl Read,
output: &mut impl Write,
username: &str,
) -> Result<(), Box<dyn std::error::Error>> {
Packet::Handshake {
username: username.into(),
}
.write(output)?;
output.flush()?;
match Packet::read_from_server(input)? {
Packet::Handshake { username: hash } if hash == "-" => {}
Packet::Handshake { .. } => {
return Err("This server requires an authenticated Minecraft session".into());
}
Packet::Disconnect { reason } => return Err(reason.into()),
_ => return Err("Invalid server handshake".into()),
}
Packet::Login {
version_or_entity: PROTOCOL_VERSION,
username: username.into(),
seed: 0,
dimension: 0,
}
.write(output)?;
output.flush()?;
Ok(())
}
fn parse_address(text: &str) -> io::Result<(String, u16)> {
let text = text.trim();
let invalid = || io::Error::new(io::ErrorKind::InvalidInput, "Invalid server address");
if text.is_empty() || text.chars().any(char::is_whitespace) || text.contains('/') {
return Err(invalid());
}
let (host, port, ipv6) = match text.strip_prefix('[').map(|rest| rest.split_once(']')) {
Some(Some((host, ""))) => (host, None, true),
Some(Some((host, port))) => (
host,
Some(port.strip_prefix(':').ok_or_else(invalid)?),
true,
),
Some(None) => return Err(invalid()),
None => match text.split_once(':') {
Some((host, port)) if !port.contains(':') => (host, Some(port), false),
_ => (text, None, text.contains(':')),
},
};
if host.is_empty() || ipv6 && host.parse::<std::net::Ipv6Addr>().is_err() {
return Err(invalid());
}
let port = port.map_or(Ok(25565), str::parse).map_err(|_| invalid())?;
Ok((host.into(), port))
}