mirror of
https://github.com/mat-1/azalea.git
synced 2025-08-02 14:26:04 +00:00
make swarm use builder
still have to fix some stuff and make client use builder
This commit is contained in:
parent
f8230e40af
commit
8940e90863
15 changed files with 206 additions and 168 deletions
1
Cargo.lock
generated
1
Cargo.lock
generated
|
|
@ -176,6 +176,7 @@ dependencies = [
|
|||
"azalea-world",
|
||||
"bevy_app",
|
||||
"bevy_ecs",
|
||||
"derive_more",
|
||||
"env_logger",
|
||||
"futures",
|
||||
"iyes_loopless",
|
||||
|
|
|
|||
|
|
@ -37,7 +37,10 @@ use azalea_world::{
|
|||
entity::Entity, EntityInfos, EntityPlugin, Local, PartialWorld, World, WorldContainer,
|
||||
};
|
||||
use bevy_app::App;
|
||||
use bevy_ecs::schedule::{IntoSystemDescriptor, Schedule, Stage, SystemSet};
|
||||
use bevy_ecs::{
|
||||
prelude::Component,
|
||||
schedule::{IntoSystemDescriptor, Schedule, Stage, SystemSet},
|
||||
};
|
||||
use bevy_time::TimePlugin;
|
||||
use iyes_loopless::prelude::*;
|
||||
use log::{debug, error};
|
||||
|
|
@ -165,11 +168,7 @@ impl Client {
|
|||
let (conn, game_profile) = Self::handshake(conn, account, address).await?;
|
||||
let (read_conn, write_conn) = conn.into_split();
|
||||
|
||||
// The buffer has to be 1 to avoid a bug where if it lags events are
|
||||
// received a bit later instead of the instant they were fired.
|
||||
// That bug especially causes issues with the pathfinder.
|
||||
let (tx, rx) = mpsc::unbounded_channel();
|
||||
tx.send(Event::Init).unwrap();
|
||||
|
||||
let mut ecs = ecs_lock.lock();
|
||||
|
||||
|
|
@ -214,8 +213,6 @@ impl Client {
|
|||
LocalPlayerEvents(tx),
|
||||
));
|
||||
|
||||
// just start up the game loop and we're ready!
|
||||
|
||||
Ok((client, rx))
|
||||
}
|
||||
|
||||
|
|
@ -366,6 +363,12 @@ impl Client {
|
|||
self.query::<&mut LocalPlayer>(ecs)
|
||||
}
|
||||
|
||||
/// Get a component from the client. This will clone the component and
|
||||
/// return it.
|
||||
pub fn component<T: Component + Clone>(&self) -> T {
|
||||
self.query::<&T>(&mut self.ecs.lock()).clone()
|
||||
}
|
||||
|
||||
/// Get a reference to our (potentially shared) world.
|
||||
///
|
||||
/// This gets the [`WeakWorld`] from our world container. If it's a normal
|
||||
|
|
|
|||
|
|
@ -33,7 +33,7 @@ impl Client {
|
|||
/// |profile: &&GameProfileComponent| profile.name == sender,
|
||||
/// );
|
||||
/// if let Some(entity) = entity {
|
||||
/// let position = bot.entity_components::<Position>(entity);
|
||||
/// let position = bot.entity_component::<Position>(entity);
|
||||
/// }
|
||||
/// ```
|
||||
pub fn entity_by<F: ReadOnlyWorldQuery, Q: ReadOnlyWorldQuery>(
|
||||
|
|
@ -45,7 +45,7 @@ impl Client {
|
|||
|
||||
/// Get a component from an entity. Note that this will return an owned type
|
||||
/// (i.e. not a reference) so it may be expensive for larger types.
|
||||
pub fn entity_components<Q: Component + Clone>(&mut self, entity: Entity) -> Q {
|
||||
pub fn entity_component<Q: Component + Clone>(&mut self, entity: Entity) -> Q {
|
||||
let mut ecs = self.ecs.lock();
|
||||
let mut q = ecs.query::<&Q>();
|
||||
let components = q
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ pub mod ping;
|
|||
mod player;
|
||||
|
||||
pub use account::Account;
|
||||
pub use bevy_ecs as ecs;
|
||||
pub use bevy_ecs;
|
||||
pub use client::{init_ecs_app, start_ecs, ChatPacket, Client, ClientInformation, JoinError};
|
||||
pub use events::Event;
|
||||
pub use local_player::{GameProfileComponent, LocalPlayer};
|
||||
|
|
|
|||
|
|
@ -14,7 +14,10 @@ use thiserror::Error;
|
|||
use tokio::{sync::mpsc, task::JoinHandle};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::{events::{Event, LocalPlayerEvents}, ClientInformation, PlayerInfo, WalkDirection};
|
||||
use crate::{
|
||||
events::{Event, LocalPlayerEvents},
|
||||
ClientInformation, PlayerInfo, WalkDirection,
|
||||
};
|
||||
|
||||
/// A player that you control that is currently in a Minecraft server.
|
||||
#[derive(Component)]
|
||||
|
|
|
|||
|
|
@ -8,32 +8,33 @@ version = "0.5.0"
|
|||
|
||||
[package.metadata.release]
|
||||
pre-release-replacements = [
|
||||
{ file = "src/lib.rs", search = "//! `azalea = \"[a-z0-9\\.-]+\"`", replace = "//! `azalea = \"{{version}}\"`" },
|
||||
{file = "src/lib.rs", search = "//! `azalea = \"[a-z0-9\\.-]+\"`", replace = "//! `azalea = \"{{version}}\"`"},
|
||||
]
|
||||
|
||||
[dependencies]
|
||||
anyhow = "^1.0.65"
|
||||
async-trait = "0.1.58"
|
||||
azalea-block = { version = "0.5.0", path = "../azalea-block" }
|
||||
azalea-chat = { version = "0.5.0", path = "../azalea-chat" }
|
||||
azalea-client = { version = "0.5.0", path = "../azalea-client" }
|
||||
azalea-core = { version = "0.5.0", path = "../azalea-core" }
|
||||
azalea-physics = { version = "0.5.0", path = "../azalea-physics" }
|
||||
azalea-protocol = { version = "0.5.0", path = "../azalea-protocol" }
|
||||
azalea-world = { version = "0.5.0", path = "../azalea-world" }
|
||||
azalea-registry = { version = "0.5.0", path = "../azalea-registry" }
|
||||
azalea-block = {version = "0.5.0", path = "../azalea-block"}
|
||||
azalea-chat = {version = "0.5.0", path = "../azalea-chat"}
|
||||
azalea-client = {version = "0.5.0", path = "../azalea-client"}
|
||||
azalea-core = {version = "0.5.0", path = "../azalea-core"}
|
||||
azalea-physics = {version = "0.5.0", path = "../azalea-physics"}
|
||||
azalea-protocol = {version = "0.5.0", path = "../azalea-protocol"}
|
||||
azalea-registry = {version = "0.5.0", path = "../azalea-registry"}
|
||||
azalea-world = {version = "0.5.0", path = "../azalea-world"}
|
||||
bevy_app = {version = "0.9.1", default-features = false}
|
||||
bevy_ecs = {version = "0.9.1", default-features = false}
|
||||
derive_more = {version = "0.99.17", features = ["deref", "deref_mut"]}
|
||||
futures = "0.3.25"
|
||||
iyes_loopless = "0.9.1"
|
||||
log = "0.4.17"
|
||||
nohash-hasher = "0.2.0"
|
||||
num-traits = "0.2.15"
|
||||
parking_lot = { version = "^0.12.1", features = ["deadlock_detection"] }
|
||||
parking_lot = {version = "^0.12.1", features = ["deadlock_detection"]}
|
||||
priority-queue = "1.3.0"
|
||||
thiserror = "^1.0.37"
|
||||
tokio = "^1.23.1"
|
||||
uuid = "1.2.2"
|
||||
bevy_app = { version = "0.9.1", default-features = false }
|
||||
bevy_ecs = { version = "0.9.1", default-features = false }
|
||||
iyes_loopless = "0.9.1"
|
||||
|
||||
[dev-dependencies]
|
||||
anyhow = "^1.0.65"
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@ async fn main() {
|
|||
.unwrap();
|
||||
}
|
||||
|
||||
#[derive(Default, Clone)]
|
||||
#[derive(Default, Clone, Component)]
|
||||
pub struct State {}
|
||||
|
||||
async fn handle(bot: Client, event: Event, _state: State) -> anyhow::Result<()> {
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
//! Automatically eat when we get hungry.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use azalea::prelude::*;
|
||||
use azalea::{Client, Event};
|
||||
use parking_lot::Mutex;
|
||||
use std::sync::Arc;
|
||||
|
|
@ -10,7 +11,7 @@ pub struct Plugin {
|
|||
pub state: State,
|
||||
}
|
||||
|
||||
#[derive(Default, Clone)]
|
||||
#[derive(Default, Clone, Component)]
|
||||
pub struct State {}
|
||||
|
||||
#[async_trait]
|
||||
|
|
|
|||
|
|
@ -22,16 +22,17 @@
|
|||
//! For faster compile times, make a `.cargo/config.toml` file in your project
|
||||
//! and copy
|
||||
//! [this file](https://github.com/mat-1/azalea/blob/main/.cargo/config.toml)
|
||||
//! into it.
|
||||
//! into it. You may have to install the LLD linker.
|
||||
//!
|
||||
//! For faster performance in debug mode, add
|
||||
//! For faster performance in debug mode, add the following code to your
|
||||
//! Cargo.toml:
|
||||
//! ```toml
|
||||
//! [profile.dev]
|
||||
//! opt-level = 1
|
||||
//! [profile.dev.package."*"]
|
||||
//! opt-level = 3
|
||||
//! ```
|
||||
//! to your Cargo.toml. You may have to install the LLD linker.
|
||||
//!
|
||||
//!
|
||||
//! # Examples
|
||||
//!
|
||||
|
|
@ -58,7 +59,7 @@
|
|||
//! .unwrap();
|
||||
//! }
|
||||
//!
|
||||
//! #[derive(Default, Clone)]
|
||||
//! #[derive(Default, Clone, Component)]
|
||||
//! pub struct State {}
|
||||
//!
|
||||
//! async fn handle(bot: Client, event: Event, state: State) -> anyhow::Result<()> {
|
||||
|
|
|
|||
|
|
@ -3,5 +3,6 @@
|
|||
|
||||
pub use crate::bot::BotClientExt;
|
||||
pub use crate::pathfinder::PathfinderClientExt;
|
||||
pub use crate::{plugins, swarm_plugins, Plugin};
|
||||
pub use azalea_client::{Account, Client, Event};
|
||||
pub use bevy_ecs;
|
||||
pub use bevy_ecs::component::Component;
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
use crate::HandleFn;
|
||||
use azalea_client::{Account, Client, Plugins};
|
||||
use azalea_client::{Account, Client};
|
||||
use azalea_protocol::ServerAddress;
|
||||
use std::future::Future;
|
||||
use thiserror::Error;
|
||||
|
|
@ -41,14 +41,6 @@ where
|
|||
pub address: A,
|
||||
/// The account that's going to join the server.
|
||||
pub account: Account,
|
||||
/// The plugins that are going to be used. Plugins are external crates that
|
||||
/// add extra functionality to Azalea. You should use the [`plugins`] macro
|
||||
/// for this field.
|
||||
///
|
||||
/// ```rust,no_run
|
||||
/// plugins![azalea_pathfinder::Plugin]
|
||||
/// ```
|
||||
pub plugins: Plugins,
|
||||
/// A struct that contains the data that you want your bot to remember
|
||||
/// across events.
|
||||
///
|
||||
|
|
@ -114,16 +106,9 @@ pub async fn start<
|
|||
|
||||
let (bot, mut rx) = Client::join(&options.account, address).await?;
|
||||
|
||||
let plugins = options.plugins;
|
||||
|
||||
let state = options.state;
|
||||
|
||||
while let Some(event) = rx.recv().await {
|
||||
let cloned_plugins = (*bot.plugins).clone();
|
||||
for plugin in cloned_plugins.into_iter() {
|
||||
tokio::spawn(plugin.handle(event.clone(), bot.clone()));
|
||||
}
|
||||
|
||||
tokio::spawn((options.handle)(bot.clone(), event.clone(), state.clone()));
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -13,24 +13,22 @@
|
|||
// in Swarm that's set to the smallest index of all the bots, and we remove all
|
||||
// messages from the queue that are before that index.
|
||||
|
||||
use crate::{Swarm, SwarmEvent};
|
||||
use async_trait::async_trait;
|
||||
use azalea_client::{packet_handling::ChatReceivedEvent, ChatPacket, Client, Event, LocalPlayer};
|
||||
use azalea_client::{packet_handling::ChatReceivedEvent, ChatPacket, LocalPlayer};
|
||||
use bevy_ecs::{
|
||||
prelude::{Component, Entity, EventReader, EventWriter},
|
||||
query::{Added, Without},
|
||||
schedule::SystemSet,
|
||||
system::{Commands, Query, Res, ResMut, Resource},
|
||||
};
|
||||
use parking_lot::Mutex;
|
||||
use std::{collections::VecDeque, sync::Arc};
|
||||
use tokio::sync::broadcast;
|
||||
use std::collections::VecDeque;
|
||||
|
||||
use crate::{Swarm, SwarmEvent};
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct Plugin;
|
||||
impl bevy_app::Plugin for Plugin {
|
||||
fn build(&self, app: &mut bevy_app::App) {
|
||||
app.add_system(chat_listener)
|
||||
app.add_event::<NewChatMessageEvent>()
|
||||
.add_system(chat_listener)
|
||||
.add_system(add_default_client_state)
|
||||
.add_system(update_min_index_and_shrink_queue)
|
||||
.insert_resource(GlobalChatState {
|
||||
|
|
@ -68,13 +66,13 @@ pub struct GlobalChatState {
|
|||
}
|
||||
|
||||
fn chat_listener(
|
||||
query: Query<&ClientChatState>,
|
||||
events: EventReader<ChatReceivedEvent>,
|
||||
mut query: Query<&mut ClientChatState>,
|
||||
mut events: EventReader<ChatReceivedEvent>,
|
||||
mut global_chat_state: ResMut<GlobalChatState>,
|
||||
new_chat_messages_events: EventWriter<NewChatMessageEvent>,
|
||||
mut new_chat_messages_events: EventWriter<NewChatMessageEvent>,
|
||||
) {
|
||||
for event in events.iter() {
|
||||
if let Ok(client_chat_state) = query.get(event.entity) {
|
||||
if let Ok(mut client_chat_state) = query.get_mut(event.entity) {
|
||||
// When a bot receives a chat messages, it looks into the queue to find the
|
||||
// earliest instance of the message content that's after the bot's chat index.
|
||||
// If it finds it, then its personal index is simply updated. Otherwise, fire
|
||||
|
|
@ -111,7 +109,7 @@ fn chat_listener(
|
|||
if !found {
|
||||
// didn't find the message, so fire the swarm event and add to the queue
|
||||
new_chat_messages_events.send(NewChatMessageEvent(event.packet.clone()));
|
||||
global_chat_state.chat_queue.push_back(event.packet);
|
||||
global_chat_state.chat_queue.push_back(event.packet.clone());
|
||||
client_chat_state.chat_index =
|
||||
global_chat_state.chat_queue.len() + global_chat_state.chat_min_index;
|
||||
}
|
||||
|
|
@ -119,38 +117,17 @@ fn chat_listener(
|
|||
}
|
||||
}
|
||||
|
||||
// impl GlobalChatState {
|
||||
// async fn start<S>(self, swarm: Swarm)
|
||||
// where
|
||||
// S: Send + Sync + Clone + 'static,
|
||||
// {
|
||||
// // it should never be locked unless we reused the same plugin for two
|
||||
// swarms // (bad)
|
||||
// let mut rx = self.rx.lock().await;
|
||||
// while let Ok(m) = rx.recv().await {
|
||||
// swarm.swarm_tx.send(SwarmEvent::Chat(m)).unwrap();
|
||||
// let bot_states = swarm
|
||||
// .bot_datas
|
||||
// .lock()
|
||||
// .iter()
|
||||
// .map(|(bot, _)| {
|
||||
// bot.plugins
|
||||
// .get::<ClientChatState>()
|
||||
// .expect("Chat plugin not installed")
|
||||
// .clone()
|
||||
// })
|
||||
// .collect::<Vec<_>>();
|
||||
// self.handle_new_chat_message(&bot_states);
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
|
||||
fn update_min_index_and_shrink_queue(
|
||||
query: Query<&ClientChatState>,
|
||||
mut global_chat_state: ResMut<GlobalChatState>,
|
||||
events: EventReader<NewChatMessageEvent>,
|
||||
mut events: EventReader<NewChatMessageEvent>,
|
||||
swarm: Res<Swarm>,
|
||||
) {
|
||||
for event in events.iter() {
|
||||
swarm
|
||||
.swarm_tx
|
||||
.send(SwarmEvent::Chat(event.0.clone()))
|
||||
.unwrap();
|
||||
// To make sure the queue doesn't grow too large, we keep a `chat_min_index`
|
||||
// in Swarm that's set to the smallest index of all the bots, and we remove all
|
||||
// messages from the queue that are before that index.
|
||||
|
|
@ -176,7 +153,7 @@ fn update_min_index_and_shrink_queue(
|
|||
}
|
||||
|
||||
// update the min index
|
||||
self.chat_min_index = new_chat_min_index;
|
||||
global_chat_state.chat_min_index = new_chat_min_index;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
43
azalea/src/swarm/events.rs
Normal file
43
azalea/src/swarm/events.rs
Normal file
|
|
@ -0,0 +1,43 @@
|
|||
use azalea_client::LocalPlayer;
|
||||
use azalea_world::entity::MinecraftEntityId;
|
||||
use bevy_ecs::{
|
||||
prelude::EventWriter,
|
||||
system::{Query, ResMut, Resource},
|
||||
};
|
||||
use derive_more::{Deref, DerefMut};
|
||||
|
||||
pub struct Plugin;
|
||||
impl bevy_app::Plugin for Plugin {
|
||||
fn build(&self, app: &mut bevy_app::App) {
|
||||
app.add_event::<SwarmReadyEvent>()
|
||||
.add_system(check_ready)
|
||||
.init_resource::<IsSwarmReady>();
|
||||
}
|
||||
}
|
||||
|
||||
/// All the bots from the swarm are now in the world.
|
||||
pub struct SwarmReadyEvent;
|
||||
|
||||
#[derive(Default, Resource, Deref, DerefMut)]
|
||||
struct IsSwarmReady(bool);
|
||||
|
||||
fn check_ready(
|
||||
query: Query<(&LocalPlayer, Option<&MinecraftEntityId>)>,
|
||||
mut is_swarm_ready: ResMut<IsSwarmReady>,
|
||||
mut ready_events: EventWriter<SwarmReadyEvent>,
|
||||
) {
|
||||
// if we already know the swarm is ready, do nothing
|
||||
if **is_swarm_ready {
|
||||
return;
|
||||
}
|
||||
// if all the players are in the world, we're ready
|
||||
for (player, entity_id) in query.iter() {
|
||||
if entity_id.is_none() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// all the players are in the world, so we're ready
|
||||
**is_swarm_ready = true;
|
||||
ready_events.send(SwarmReadyEvent);
|
||||
}
|
||||
|
|
@ -1,5 +1,7 @@
|
|||
/// Swarms are a way to conveniently control many bots.
|
||||
//! Swarms are a way to conveniently control many bots.
|
||||
|
||||
mod chat;
|
||||
mod events;
|
||||
|
||||
use crate::HandleFn;
|
||||
use azalea_client::{init_ecs_app, start_ecs, Account, ChatPacket, Client, Event, JoinError};
|
||||
|
|
@ -10,7 +12,7 @@ use azalea_protocol::{
|
|||
};
|
||||
use azalea_world::WorldContainer;
|
||||
use bevy_app::Plugin;
|
||||
use bevy_ecs::system::Resource;
|
||||
use bevy_ecs::{prelude::Component, system::Resource};
|
||||
use futures::future::join_all;
|
||||
use log::error;
|
||||
use parking_lot::{Mutex, RwLock};
|
||||
|
|
@ -29,12 +31,14 @@ use tokio::sync::mpsc;
|
|||
/// [`azalea::start_swarm`]: fn.start_swarm.html
|
||||
#[derive(Clone, Resource)]
|
||||
pub struct Swarm {
|
||||
pub ecs: Arc<Mutex<bevy_ecs::world::World>>,
|
||||
pub ecs_lock: Arc<Mutex<bevy_ecs::world::World>>,
|
||||
|
||||
bots: Arc<Mutex<Vec<Client>>>,
|
||||
|
||||
// bot_datas: Arc<Mutex<Vec<(Client, S)>>>,
|
||||
resolved_address: SocketAddr,
|
||||
address: ServerAddress,
|
||||
pub world_container: Arc<RwLock<WorldContainer>>,
|
||||
pub ecs_lock: Arc<Mutex<bevy_ecs::world::World>>,
|
||||
|
||||
bots_tx: mpsc::UnboundedSender<(Option<Event>, Client)>,
|
||||
swarm_tx: mpsc::UnboundedSender<SwarmEvent>,
|
||||
|
|
@ -72,10 +76,10 @@ where
|
|||
}
|
||||
impl<S, SS, Fut, SwarmFut> SwarmBuilder<S, SS, Fut, SwarmFut>
|
||||
where
|
||||
Fut: Future<Output = Result<(), anyhow::Error>>,
|
||||
SwarmFut: Future<Output = Result<(), anyhow::Error>>,
|
||||
S: Default + Send + Sync + Clone + 'static,
|
||||
SS: Default + Send + Sync + Clone + 'static,
|
||||
Fut: Future<Output = Result<(), anyhow::Error>> + Send + 'static,
|
||||
SwarmFut: Future<Output = Result<(), anyhow::Error>> + Send + 'static,
|
||||
S: Default + Send + Sync + Clone + Component + 'static,
|
||||
SS: Default + Send + Sync + Clone + Component + 'static,
|
||||
{
|
||||
/// Start creating the swarm.
|
||||
pub fn new() -> Self {
|
||||
|
|
@ -91,6 +95,7 @@ where
|
|||
swarm_handler: None,
|
||||
join_delay: None,
|
||||
}
|
||||
.add_default_swarm_plugins()
|
||||
}
|
||||
|
||||
/// Add a vec of [`Account`]s to the swarm.
|
||||
|
|
@ -98,25 +103,26 @@ where
|
|||
/// Use [`Self::add_account`] to only add one account. If you want the
|
||||
/// clients to have different default states, add them one at a time with
|
||||
/// [`Self::add_account_with_state`].
|
||||
pub fn add_accounts(&mut self, accounts: Vec<Account>) {
|
||||
pub fn add_accounts(mut self, accounts: Vec<Account>) -> Self {
|
||||
for account in accounts {
|
||||
self.add_account(account);
|
||||
self = self.add_account(account);
|
||||
}
|
||||
self
|
||||
}
|
||||
/// Add a single new [`Account`] to the swarm. Use [`add_accounts`] to add
|
||||
/// multiple accounts at a time.
|
||||
///
|
||||
/// This will make the state for this client be the default, use
|
||||
/// [`Self::add_account_with_state`] to avoid that.
|
||||
pub fn add_account(&mut self, account: Account) {
|
||||
self.accounts.push(account);
|
||||
self.states.push(S::default());
|
||||
pub fn add_account(mut self, account: Account) -> Self {
|
||||
self.add_account_with_state(account, S::default())
|
||||
}
|
||||
/// Add an account with a custom initial state. Use just
|
||||
/// [`Self::add_account`] to use the Default implementation for the state.
|
||||
pub fn add_account_with_state(&mut self, account: Account, state: S) {
|
||||
pub fn add_account_with_state(mut self, account: Account, state: S) -> Self {
|
||||
self.accounts.push(account);
|
||||
self.states.push(state);
|
||||
self
|
||||
}
|
||||
|
||||
/// Set the function that's called every time a bot receives an [`Event`].
|
||||
|
|
@ -125,8 +131,9 @@ where
|
|||
/// You can only have one client handler, calling this again will replace
|
||||
/// the old client handler function (you can have a client handler and swarm
|
||||
/// handler separately though).
|
||||
pub fn set_handler(&mut self, handler: HandleFn<Fut, S>) {
|
||||
pub fn set_handler(mut self, handler: HandleFn<Fut, S>) -> Self {
|
||||
self.handler = Some(handler);
|
||||
self
|
||||
}
|
||||
/// Set the function that's called every time the swarm receives a
|
||||
/// [`SwarmEvent`]. This is the way to handle global swarm events.
|
||||
|
|
@ -134,15 +141,32 @@ where
|
|||
/// You can only have one swarm handler, calling this again will replace
|
||||
/// the old swarm handler function (you can have a client handler and swarm
|
||||
/// handler separately though).
|
||||
pub fn set_swarm_handler(&mut self, handler: SwarmHandleFn<SwarmFut, SS>) {
|
||||
pub fn set_swarm_handler(mut self, handler: SwarmHandleFn<SwarmFut, SS>) -> Self {
|
||||
self.swarm_handler = Some(handler);
|
||||
self
|
||||
}
|
||||
|
||||
/// TODO: write plugin docs probably here
|
||||
///
|
||||
/// notes: ECS entities might not be Minecraft entities, filter by MinecraftEntityId to make sure
|
||||
fn add_plugin<T: Plugin>(&mut self, plugin: T) {
|
||||
///
|
||||
/// notes: ECS entities might not be Minecraft entities, filter by
|
||||
/// MinecraftEntityId to make sure
|
||||
pub fn add_plugin<T: Plugin>(mut self, plugin: T) -> Self {
|
||||
self.app.add_plugin(plugin);
|
||||
self
|
||||
}
|
||||
|
||||
/// Set how long we should wait between each bot joining the server.
|
||||
///
|
||||
/// By default, every bot will connect at the same time. If you set this
|
||||
/// field, however, the bots will wait for the previous one to have
|
||||
/// connected and *then* they'll wait the given duration.
|
||||
pub fn join_delay(mut self, delay: std::time::Duration) -> Self {
|
||||
self.join_delay = Some(delay);
|
||||
self
|
||||
}
|
||||
|
||||
fn add_default_swarm_plugins(self) -> Self {
|
||||
self.add_plugin(chat::Plugin).add_plugin(events::Plugin)
|
||||
}
|
||||
|
||||
/// Build this `SwarmBuilder` into an actual [`Swarm`] and join the given
|
||||
|
|
@ -152,7 +176,7 @@ where
|
|||
/// that implements `TryInto<ServerAddress>`.
|
||||
///
|
||||
/// [`ServerAddress`]: azalea_protocol::ServerAddress
|
||||
pub async fn start(self, address: impl TryInto<ServerAddress>) -> Result<(), SwarmStartError>{
|
||||
pub async fn start(self, address: impl TryInto<ServerAddress>) -> Result<(), SwarmStartError> {
|
||||
assert_eq!(
|
||||
self.accounts.len(),
|
||||
self.states.len(),
|
||||
|
|
@ -177,8 +201,10 @@ where
|
|||
let (run_schedule_sender, run_schedule_receiver) = mpsc::channel(1);
|
||||
let ecs_lock = start_ecs(self.app, run_schedule_receiver, run_schedule_sender.clone());
|
||||
|
||||
let mut swarm = Swarm {
|
||||
// bot_datas: Arc::new(Mutex::new(Vec::new())),
|
||||
let swarm = Swarm {
|
||||
ecs_lock: ecs_lock.clone(),
|
||||
bots: Arc::new(Mutex::new(Vec::new())),
|
||||
|
||||
resolved_address,
|
||||
address,
|
||||
world_container,
|
||||
|
|
@ -187,26 +213,30 @@ where
|
|||
|
||||
swarm_tx: swarm_tx.clone(),
|
||||
|
||||
ecs_lock,
|
||||
run_schedule_sender,
|
||||
};
|
||||
self.app.insert_resource(swarm.clone());
|
||||
|
||||
swarm.plugins.add(chat::Plugin);
|
||||
ecs_lock.lock().insert_resource(swarm.clone());
|
||||
|
||||
// SwarmBuilder (self) isn't Send so we have to take all the things we need out
|
||||
// of it
|
||||
let mut swarm_clone = swarm.clone();
|
||||
let join_delay = self.join_delay.clone();
|
||||
let accounts = self.accounts.clone();
|
||||
let states = self.states.clone();
|
||||
|
||||
let join_task = tokio::spawn(async move {
|
||||
if let Some(join_delay) = self.join_delay {
|
||||
if let Some(join_delay) = join_delay {
|
||||
// if there's a join delay, then join one by one
|
||||
for (account, state) in self.accounts.iter().zip(self.states) {
|
||||
for (account, state) in accounts.iter().zip(states) {
|
||||
swarm_clone
|
||||
.add_with_exponential_backoff(account, state.clone())
|
||||
.await;
|
||||
tokio::time::sleep(join_delay).await;
|
||||
}
|
||||
} else {
|
||||
// otherwise, join all at once
|
||||
let swarm_borrow = &swarm_clone;
|
||||
join_all(self.accounts.iter().zip(self.states).map(
|
||||
join_all(accounts.iter().zip(states).map(
|
||||
async move |(account, state)| -> Result<(), JoinError> {
|
||||
swarm_borrow
|
||||
.clone()
|
||||
|
|
@ -220,35 +250,27 @@ where
|
|||
});
|
||||
|
||||
let swarm_state = self.swarm_state;
|
||||
let mut internal_state = InternalSwarmState::default();
|
||||
|
||||
// Watch swarm_rx and send those events to the swarm_handle.
|
||||
let swarm_clone = swarm.clone();
|
||||
tokio::spawn(async move {
|
||||
while let Some(event) = swarm_rx.recv().await {
|
||||
tokio::spawn((self.swarm_handle)(
|
||||
swarm_clone.clone(),
|
||||
event,
|
||||
swarm_state.clone(),
|
||||
));
|
||||
if let Some(swarm_handler) = self.swarm_handler {
|
||||
tokio::spawn((swarm_handler)(
|
||||
swarm_clone.clone(),
|
||||
event,
|
||||
swarm_state.clone(),
|
||||
));
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// bot events
|
||||
while let Some((Some(event), (bot, state))) = bots_rx.recv().await {
|
||||
// remove this #[allow] when more checks are added
|
||||
// TODO: actually it'd be better to just have this in a system
|
||||
#[allow(clippy::single_match)]
|
||||
match &event {
|
||||
Event::Login => {
|
||||
internal_state.bots_joined += 1;
|
||||
if internal_state.bots_joined == swarm.bot_datas.lock().len() {
|
||||
swarm_tx.send(SwarmEvent::Login).unwrap();
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
while let Some((Some(event), bot)) = bots_rx.recv().await {
|
||||
if let Some(handler) = self.handler {
|
||||
let state = bot.component::<S>();
|
||||
tokio::spawn((handler)(bot, event, state));
|
||||
}
|
||||
tokio::spawn((self.handle)(bot, event, state));
|
||||
}
|
||||
|
||||
join_task.abort();
|
||||
|
|
@ -372,7 +394,11 @@ pub enum SwarmStartError {
|
|||
impl Swarm {
|
||||
/// Add a new account to the swarm. You can remove it later by calling
|
||||
/// [`Client::disconnect`].
|
||||
pub async fn add<S: Clone>(&mut self, account: &Account, state: S) -> Result<Client, JoinError> {
|
||||
pub async fn add<S: Component + Clone>(
|
||||
&mut self,
|
||||
account: &Account,
|
||||
state: S,
|
||||
) -> Result<Client, JoinError> {
|
||||
// tx is moved to the bot so it can send us events
|
||||
// rx is used to receive events from the bot
|
||||
// An event that causes the schedule to run. This is only used internally.
|
||||
|
|
@ -386,10 +412,14 @@ impl Swarm {
|
|||
self.run_schedule_sender.clone(),
|
||||
)
|
||||
.await?;
|
||||
// add the state to the client
|
||||
{
|
||||
let mut ecs = self.ecs_lock.lock();
|
||||
ecs.entity_mut(bot.entity).insert(state);
|
||||
}
|
||||
|
||||
let cloned_bots_tx = self.bots_tx.clone();
|
||||
let cloned_bot = bot.clone();
|
||||
let cloned_state = state.clone();
|
||||
let owned_account = account.clone();
|
||||
let swarm_tx = self.swarm_tx.clone();
|
||||
// send the init event immediately so it's the first thing we get
|
||||
|
|
@ -398,9 +428,7 @@ impl Swarm {
|
|||
while let Some(event) = rx.recv().await {
|
||||
// we can't handle events here (since we can't copy the handler),
|
||||
// they're handled above in start_swarm
|
||||
if let Err(e) =
|
||||
cloned_bots_tx.send((Some(event), (cloned_bot.clone(), cloned_state.clone())))
|
||||
{
|
||||
if let Err(e) = cloned_bots_tx.send((Some(event), cloned_bot.clone())) {
|
||||
error!("Error sending event to swarm: {e}");
|
||||
}
|
||||
}
|
||||
|
|
@ -409,7 +437,6 @@ impl Swarm {
|
|||
.unwrap();
|
||||
});
|
||||
|
||||
|
||||
Ok(bot)
|
||||
}
|
||||
|
||||
|
|
@ -418,7 +445,11 @@ impl Swarm {
|
|||
///
|
||||
/// Exponential backoff means if it fails joining it will initially wait 10
|
||||
/// seconds, then 20, then 40, up to 2 minutes.
|
||||
pub async fn add_with_exponential_backoff(&mut self, account: &Account, state: S) -> Client {
|
||||
pub async fn add_with_exponential_backoff<S: Component + Clone>(
|
||||
&mut self,
|
||||
account: &Account,
|
||||
state: S,
|
||||
) -> Client {
|
||||
let mut disconnects = 0;
|
||||
loop {
|
||||
match self.add(account, state.clone()).await {
|
||||
|
|
@ -436,32 +467,23 @@ impl Swarm {
|
|||
}
|
||||
}
|
||||
|
||||
impl IntoIterator for Swarm
|
||||
where
|
||||
S: Send + Sync + Clone + 'static,
|
||||
{
|
||||
type Item = (Client, S);
|
||||
impl IntoIterator for Swarm {
|
||||
type Item = Client;
|
||||
type IntoIter = std::vec::IntoIter<Self::Item>;
|
||||
|
||||
/// Iterate over the bots and their states in this swarm.
|
||||
/// Iterate over the bots in this swarm.
|
||||
///
|
||||
/// ```rust,no_run
|
||||
/// for (bot, state) in swarm {
|
||||
/// for bot in swarm {
|
||||
/// let state = bot.component::<State>();
|
||||
/// // ...
|
||||
/// }
|
||||
/// ```
|
||||
fn into_iter(self) -> Self::IntoIter {
|
||||
self.
|
||||
// self.bot_datas.lock().clone().into_iter()
|
||||
self.bots.lock().clone().into_iter()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct InternalSwarmState {
|
||||
/// The number of bots connected to the server
|
||||
pub bots_joined: usize,
|
||||
}
|
||||
|
||||
impl From<ConnectionError> for SwarmStartError {
|
||||
fn from(e: ConnectionError) -> Self {
|
||||
SwarmStartError::from(JoinError::from(e))
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
#![feature(type_alias_impl_trait)]
|
||||
|
||||
use azalea::ecs::query::With;
|
||||
use azalea::bevy_ecs::query::With;
|
||||
use azalea::entity::metadata::Player;
|
||||
use azalea::entity::Position;
|
||||
use azalea::pathfinder::BlockPosGoal;
|
||||
|
|
@ -10,10 +10,10 @@ use azalea::{Account, Client, Event};
|
|||
use azalea_protocol::packets::game::serverbound_client_command_packet::ServerboundClientCommandPacket;
|
||||
use std::time::Duration;
|
||||
|
||||
#[derive(Default, Clone)]
|
||||
#[derive(Default, Clone, Component)]
|
||||
struct State {}
|
||||
|
||||
#[derive(Default, Clone)]
|
||||
#[derive(Default, Clone, Component)]
|
||||
struct SwarmState {}
|
||||
|
||||
#[tokio::main]
|
||||
|
|
@ -56,8 +56,8 @@ async fn main() -> anyhow::Result<()> {
|
|||
let e = azalea::SwarmBuilder::new()
|
||||
.add_accounts(accounts.clone())
|
||||
.set_handler(handle)
|
||||
.swarm_handle(swarm_handle)
|
||||
.join_delay(Some(Duration::from_millis(1000)))
|
||||
.set_swarm_handler(swarm_handle)
|
||||
.join_delay(Duration::from_millis(1000))
|
||||
.start("localhost")
|
||||
.await;
|
||||
println!("{e:?}");
|
||||
|
|
@ -99,12 +99,12 @@ async fn handle(mut bot: Client, event: Event, _state: State) -> anyhow::Result<
|
|||
if let Some(entity) = entity {
|
||||
match m.content().as_str() {
|
||||
"goto" => {
|
||||
let entity_pos = bot.entity_components::<Position>(entity);
|
||||
let entity_pos = bot.entity_component::<Position>(entity);
|
||||
let target_pos: BlockPos = entity_pos.into();
|
||||
bot.goto(BlockPosGoal::from(target_pos));
|
||||
}
|
||||
"look" => {
|
||||
let entity_pos = bot.entity_components::<Position>(entity);
|
||||
let entity_pos = bot.entity_component::<Position>(entity);
|
||||
let target_pos: BlockPos = entity_pos.into();
|
||||
println!("target_pos: {:?}", target_pos);
|
||||
bot.look_at(target_pos.center());
|
||||
|
|
@ -138,7 +138,7 @@ async fn handle(mut bot: Client, event: Event, _state: State) -> anyhow::Result<
|
|||
}
|
||||
|
||||
async fn swarm_handle(
|
||||
mut swarm: Swarm<State>,
|
||||
mut swarm: Swarm,
|
||||
event: SwarmEvent,
|
||||
_state: SwarmState,
|
||||
) -> anyhow::Result<()> {
|
||||
|
|
@ -151,7 +151,7 @@ async fn swarm_handle(
|
|||
SwarmEvent::Chat(m) => {
|
||||
println!("swarm chat message: {}", m.message().to_ansi());
|
||||
if m.message().to_string() == "<py5> world" {
|
||||
for (name, world) in &swarm.worlds.read().worlds {
|
||||
for (name, world) in &swarm.world_container.read().worlds {
|
||||
println!("world name: {}", name);
|
||||
if let Some(w) = world.upgrade() {
|
||||
for chunk_pos in w.read().chunks.chunks.values() {
|
||||
|
|
@ -163,7 +163,7 @@ async fn swarm_handle(
|
|||
}
|
||||
}
|
||||
if m.message().to_string() == "<py5> hi" {
|
||||
for (bot, _) in swarm {
|
||||
for bot in swarm {
|
||||
bot.chat("hello");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue