use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc::{channel, Receiver};
use std::sync::{Arc, Mutex, MutexGuard};
use fnv::FnvHashMap;
use futures::future::{self, Executor};
use futures::sync::mpsc;
use futures::{Future, Stream};
use hibitset::{BitSet, BitSetLike};
use super::{Asset, Event, Export, Exporter, Handle, Import, Importer, Storage};
pub struct Manager<A, L, S>
where
A: Asset,
L: Importer<Output = <A as Import>::Input>,
S: Exporter<Input = <A as Export>::Output>,
{
id: AtomicUsize,
event_sender: mpsc::UnboundedSender<Event<A, L, S>>,
receiver: Receiver<Event<A, L, S>>,
storage: Arc<Storage<A>>,
saving: Arc<Mutex<BitSet>>,
importing: Arc<Mutex<BitSet>>,
imported: Arc<Mutex<BitSet>>,
importer: Arc<L>,
exporter: Arc<S>,
data: FnvHashMap<
Handle<A>,
(
L::Path,
L::Options,
<A as Import>::Options,
S::Path,
S::Options,
<A as Export>::Options,
),
>,
executor: Arc<Executor<Box<Future<Item = (), Error = ()> + Send>>>,
}
unsafe impl<A, L, S> Send for Manager<A, L, S>
where
A: Asset,
L: Importer<Output = <A as Import>::Input>,
S: Exporter<Input = <A as Export>::Output>,
{}
unsafe impl<A, L, S> Sync for Manager<A, L, S>
where
A: Asset,
L: Importer<Output = <A as Import>::Input>,
S: Exporter<Input = <A as Export>::Output>,
{}
impl<A, L, S> Manager<A, L, S>
where
A: Asset,
L: 'static + Importer<Output = <A as Import>::Input>,
S: 'static + Exporter<Input = <A as Export>::Output>,
{
#[inline]
pub fn new(
importer: Arc<L>,
exporter: Arc<S>,
executor: Arc<Executor<Box<Future<Item = (), Error = ()> + Send>>>,
) -> Self {
let (sender, receiver) = channel();
let (event_sender, event_receiver) = mpsc::unbounded();
let future_receiver = event_receiver.map_err(|_| ()).for_each(move |event| {
sender.send(event).unwrap();
Ok(())
});
executor.execute(Box::new(future_receiver)).unwrap();
Manager {
id: AtomicUsize::new(0),
event_sender: event_sender,
receiver: receiver,
storage: Arc::new(Storage::default()),
saving: Arc::new(Mutex::new(BitSet::default())),
importing: Arc::new(Mutex::new(BitSet::default())),
imported: Arc::new(Mutex::new(BitSet::default())),
data: FnvHashMap::default(),
importer: importer,
exporter: exporter,
executor: executor.clone(),
}
}
#[inline]
pub fn add(
&mut self,
import_path: L::Path,
importer_options: L::Options,
import_options: <A as Import>::Options,
export_path: S::Path,
exporter_options: S::Options,
export_options: <A as Export>::Options,
preimport: bool,
) -> Handle<A> {
let id = self.next_id();
let handle = Handle::new(id);
unsafe {
self.storage.reserve(id as usize);
}
self.data.insert(
handle,
(
import_path,
importer_options,
import_options,
export_path,
exporter_options,
export_options,
),
);
if preimport {
self.get(&handle);
}
handle
}
#[inline]
pub fn get(&self, handle: &Handle<A>) -> Option<Arc<A>> {
let id = handle.id();
let imported = self.imported();
let mut importing = self.importing();
if !importing.contains(id) {
println!("get {:?}", id);
importing.add(id);
self.import(*handle);
None
} else if !imported.contains(id) {
None
} else {
self.storage.get(id)
}
}
#[inline]
pub fn reimport(&self, handle: &Handle<A>) {
let id = handle.id();
let handle = *handle;
let mut importing = self.importing();
if !importing.contains(id) {
importing.add(id);
let event_sender = self.event_sender.clone();
let f = self.import_future(handle).map(move |_| {
event_sender
.unbounded_send(Event::Reimport(handle))
.expect("failed to send asset reimport event")
});
self.executor.execute(Box::new(f)).unwrap();
}
}
#[inline]
pub fn export(&self, handle: &Handle<A>) {
let id = handle.id();
let mut saving = self.saving();
let imported = self.imported();
if imported.contains(id) && !saving.contains(id) {
saving.add(id);
let f = self.export_future(*handle);
self.executor.execute(Box::new(f)).unwrap();
}
}
#[inline]
fn import(&self, handle: Handle<A>) {
let f = self.import_future(handle);
self.executor.execute(Box::new(f)).unwrap();
}
#[inline]
fn import_future(&self, handle: Handle<A>) -> impl Future<Item = (), Error = ()> {
let id = handle.id();
let &(ref p, ref lo, ref io, _, _, _) = self
.data
.get(&handle)
.expect("import called with invalid id");
let path = p.clone();
let importer_options = lo.clone();
let import_options = io.clone();
let storage = self.storage.clone();
let importing = self.importing.clone();
let imported = self.imported.clone();
let import_event_sender = self.event_sender.clone();
let importer_event_sender = self.event_sender.clone();
self.importer
.import(path, importer_options)
.map(move |output| {
importing
.lock()
.expect("failed to acquire importing lock")
.remove(id);
match <A as Import>::import(output, import_options) {
Ok(a) => {
unsafe {
storage.insert(id, a);
}
imported
.lock()
.expect("failed to acquire imported lock")
.add(id);
import_event_sender
.unbounded_send(Event::Load(handle))
.expect("failed to send asset import event")
}
Err(error) => import_event_sender
.unbounded_send(Event::ImportError(handle, error))
.expect("failed to send asset import error event"),
}
})
.map_err(move |error| {
importer_event_sender
.unbounded_send(Event::LoadError(handle, error))
.expect("failed to send asset import error event")
})
}
#[inline]
fn export_future(&self, handle: Handle<A>) -> impl Future<Item = (), Error = ()> {
let id = handle.id();
let &(_, _, _, ref p, ref so, ref eo) = self
.data
.get(&handle)
.expect("import called with invalid id");
let path = p.clone();
let exporter_options = so.clone();
let export_options = eo.clone();
let exporter = self.exporter.clone();
let saving = self.importing.clone();
let exporter_event_sender = self.event_sender.clone();
let exporter_error_event_sender = self.event_sender.clone();
let export_event_sender = self.event_sender.clone();
let input = self
.storage
.get(id)
.expect("failed to get asset from storage");
future::lazy(move || future::result(<A as Export>::export(&*input, export_options)))
.map_err(move |error| {
export_event_sender
.unbounded_send(Event::ExportError(handle, error))
.expect("failed to send asset export error event")
})
.and_then(move |output| {
exporter
.export(output, path, exporter_options)
.map(move |_| {
saving
.lock()
.expect("failed to acquire importing lock")
.remove(id);
exporter_event_sender
.unbounded_send(Event::Save(handle))
.expect("failed to send asset export event")
})
.map_err(move |error| {
exporter_error_event_sender
.unbounded_send(Event::SaveError(handle, error))
.expect("failed to send asset export error event")
})
})
}
}
impl<A, L, S> Manager<A, L, S>
where
A: Asset,
L: Importer<Output = <A as Import>::Input>,
S: Exporter<Input = <A as Export>::Output>,
{
#[inline]
pub fn wait_event(&self) -> Option<Event<A, L, S>> {
match self.receiver.recv() {
Ok(event) => Some(event),
Err(_) => None,
}
}
#[inline]
pub fn poll_event(&self) -> Option<Event<A, L, S>> {
match self.receiver.try_recv() {
Ok(event) => Some(event),
Err(_) => None,
}
}
#[inline]
pub fn poll_events(&self) -> Vec<Event<A, L, S>> {
let mut events = Vec::new();
while let Ok(event) = self.receiver.try_recv() {
events.push(event);
}
events
}
#[inline]
fn next_id(&self) -> u32 {
self.id.fetch_add(1, Ordering::Relaxed) as u32
}
#[inline]
fn saving(&self) -> MutexGuard<BitSet> {
self.saving.lock().expect("failed to acquire saving lock")
}
#[inline]
fn importing(&self) -> MutexGuard<BitSet> {
self.importing
.lock()
.expect("failed to acquire importing lock")
}
#[inline]
fn imported(&self) -> MutexGuard<BitSet> {
self.imported
.lock()
.expect("failed to acquire imported lock")
}
#[inline(always)]
pub fn data(
&self,
) -> &FnvHashMap<
Handle<A>,
(
L::Path,
L::Options,
<A as Import>::Options,
S::Path,
S::Options,
<A as Export>::Options,
),
> {
&self.data
}
#[inline]
pub fn is_imported(&self, handle: &Handle<A>) -> bool {
self.imported().contains(handle.id())
}
#[inline]
pub fn is_importing(&self, handle: &Handle<A>) -> bool {
self.importing().contains(handle.id())
}
#[inline]
pub fn count(&self) -> usize {
self.data.len()
}
#[inline]
pub fn imported_count(&self) -> usize {
let imported = self.imported();
(&*imported).iter().count()
}
#[inline]
pub fn importing_count(&self) -> usize {
let importing = self.importing();
(&*importing).iter().count()
}
#[inline]
pub fn remove(&mut self, handle: Handle<A>) -> bool {
let id = handle.id();
if self.imported().remove(id) {
let asset = unsafe { self.storage.remove(id) };
self.event_sender
.unbounded_send(Event::Remove(handle, asset))
.expect("failed to send asset event");
true
} else {
false
}
}
}
impl<A, L, S> Drop for Manager<A, L, S>
where
A: Asset,
L: Importer<Output = <A as Import>::Input>,
S: Exporter<Input = <A as Export>::Output>,
{
#[inline]
fn drop(&mut self) {
unsafe {
self.storage.clean(&*self.imported());
}
}
}