otter/
mgmtchannel.rs

1// Copyright 2020-2021 Ian Jackson and contributors to Otter
2// SPDX-License-Identifier: AGPL-3.0-or-later
3// There is NO WARRANTY.
4
5use crate::prelude::*;
6use crate::commands::*;
7
8pub struct MgmtChannel<R:Read, W:Write> {
9  pub read:  FrameReader<R>,
10  pub write: FrameWriter<W>,
11}
12
13impl<R,W> Debug for MgmtChannel<R,W> where R: Read, W: Write {
14  #[throws(fmt::Error)]
15  fn fmt(&self, f: &mut fmt::Formatter) {
16    f.write_str("MgmtChannel{...}")?
17  }
18}
19
20pub type ClientMgmtChannel = MgmtChannel<
21    Box<dyn ReadDebug  + Send + 'static>,
22    Box<dyn WriteDebug + Send + 'static>,
23  >;
24
25pub trait ReadDebug: Read + Debug { }
26pub trait WriteDebug: Write + Debug { }
27impl<T> ReadDebug for T where T: Read + Debug { }
28impl<T> WriteDebug for T where T: Write + Debug { }
29
30impl ClientMgmtChannel {
31  #[throws(AE)]
32  pub fn connect(socket_path: &str) -> Self {
33    let unix = UnixStream::connect(socket_path)
34      .with_context(||socket_path.to_owned())
35      .context("connect to server")?; 
36    let read = unix.try_clone().context("dup the client connection")?;
37    let write = unix;
38    MgmtChannel::new_boxed(read, write)
39  }
40
41  pub fn new_boxed<R,W>(read: R, write: W) -> Self
42  where R: ReadDebug  + Send + 'static,
43        W: WriteDebug + Send + 'static {
44    MgmtChannel::new_raw(Box::new(read), Box::new(write))
45  }
46}
47
48impl MgmtChannel<TimedFdReader,TimedFdWriter> {
49  #[throws(AE)]
50  pub fn new_timed<U>(conn: U) -> Self
51  where U: IoTryClone + Read + Write + IntoRawFd + Send + 'static,
52  {
53    let read = conn.try_clone().context("dup the command stream")?;
54    let read = TimedFdReader::new(read).context("set up timed reader")?;
55    let write = TimedFdWriter::new(conn).context("set up timed writerr")?;
56    MgmtChannel::new_raw(read, write)
57  }
58}
59
60impl<R,W> MgmtChannel<R,W> where R: Read, W: Write + Send {
61  pub fn new_raw(read: R, write: W) -> Self  {
62    let read = FrameReader::new(read);
63    let write = FrameWriter::new(write);
64    MgmtChannel { read, write }
65  }
66
67  pub fn read_inner_mut(&mut self) -> &mut R {
68    self.read.inner_mut()
69  }
70}
71
72impl ClientMgmtChannel {
73  pub const PROGRESS: ProgressUpdateMode = PUM::Duplex;
74
75  #[throws(AE)]
76  pub fn cmd_withbulk<U,D>(&mut self, cmd: &MgmtCommand,
77                           up: &mut U, down: &mut D,
78                           progress: &mut dyn termprogress::Reporter)
79                           -> MgmtResponse
80  where U: Read + Send, D: Write,
81  {
82    use MgmtResponse::*;
83    let mut wbulk = self.write
84      .write_withbulk().context("start sending command")?
85      .respond(&cmd).context("send command")?;
86    let read = &mut self.read;
87
88    let (resp, mut rbulk) = crossbeam_utils::thread::scope(|scope| {
89      let thr = scope.spawn(move |_| {
90        io::copy(up, &mut wbulk).context("copy")?;
91        wbulk.finish().context("finish")?;
92        Ok::<_,AE>(())
93      });
94
95      let (mut resp, mut rbulk) =
96        read.read_withbulk()
97        .context("failed to read response")?;
98
99      while let MR::Progress(pi) = resp {
100        resp = (&mut rbulk).read_rmp()?;
101        progress.report(&pi);
102      }
103
104      let r = thr.join().expect("bulk data upload thread paniced");
105      if let Err(e) = r {
106        progress.clear();
107        warn!("bulk data upload failed: {}", e);
108      }
109      Ok::<_,AE>((resp, rbulk))
110    })
111      .expect("bulk data upload thread panicked, not reaped")
112      ?;
113
114    progress.clear();
115    match &resp {
116      Progress(_) => panic!(),
117      Fine | AccountsList{..} | GamesList{..} |
118      Libraries(_) | LibraryItems(_) | Bundles{..} | Bundle{..} => { },
119      SshKeys(..) | SshKeyAdded{..} | ThisConnAuthBy{..} => { },
120      AlterGame { error: None, .. } => { },
121      Error { error } => {
122        Err(error.clone()).context(
123          format!("got error response to: {:?}",&cmd)
124        )?;
125      },
126      AlterGame { error: Some(error), ref responses } => {
127        if let MgmtCommand::AlterGame { insns, .. } = &cmd {
128          if responses.len() < insns.len() {
129            Err(error.clone())
130              .with_context(|| format!("{:?}", &insns[responses.len()]))
131              .context("AlterGame insn failed")?;
132          }
133        }
134        Err(error.clone()).context(format!(
135          "game alterations failed (maybe partially); response to: {:?}",
136          &cmd
137        ))?;
138      }
139    };
140
141    io::copy(&mut rbulk, down).context("copy bulk download")?;
142    resp
143  }
144
145  #[throws(AE)]
146  pub fn cmd(&mut self, cmd: &MgmtCommand) -> MgmtResponse {
147    self.cmd_withbulk(cmd, &mut io::empty(), &mut io::sink(),
148                      &mut termprogress::Null)?
149  }
150
151  pub fn for_game(self, game: InstanceName, how: MgmtGameUpdateMode)
152                  -> MgmtChannelForGame {
153    MgmtChannelForGame {
154      chan: self,
155      game, how
156    }
157  }
158}
159
160pub trait IoTryClone: Sized {
161  fn try_clone(&self) -> io::Result<Self>;
162}
163
164impl IoTryClone for UnixStream {
165  fn try_clone(&self) -> io::Result<UnixStream> { self.try_clone() }
166}
167
168
169#[derive(Debug,Deref,DerefMut)]
170pub struct MgmtChannelForGame {
171  #[deref] #[deref_mut] pub chan: ClientMgmtChannel,
172  pub game: InstanceName,
173  pub how: MgmtGameUpdateMode,
174}
175
176impl MgmtChannelForGame {
177  #[throws(AE)]
178  pub fn alter_game(&mut self, insns: Vec<MgmtGameInstruction>,
179                f: Option<&mut dyn FnMut(&MgmtGameResponse) -> Result<(),AE>>)
180                -> Vec<MgmtGameResponse> {
181    let insns_len = insns.len();
182    let cmd = MgmtCommand::AlterGame {
183      game: self.game.clone(), how: self.how,
184      insns
185    };
186    let responses = match self.cmd(&cmd)? {
187      MgmtResponse::AlterGame { error: None, responses }
188      if responses.len() == insns_len ||
189         responses.iter().any(|r| matches!(r, MGR::InsnExpanded)) => {
190        responses
191      },
192      wat => Err(anyhow!("unexpected AlterGame response: {:?} => {:?}",
193                         &cmd, &wat))?,
194    };
195    if let Some(f) = f {
196      for response in &responses {
197        f(response)?;
198      }
199    }
200    responses
201  }
202
203  #[throws(AE)]
204  pub fn info(&mut self) -> MgmtGameResponseGameInfo {
205    let resp = self.alter_game(vec![MGI::Info], None)?;
206    match &resp[..] {
207      [MGR::Info(info)] => info.clone(),
208      x => throw!(anyhow!("unexpected response to game Info: {:?}", &x)),
209    }
210  }
211
212  #[throws(AE)]
213  pub fn has_player(&mut self, account: &AccountName)
214                    -> Option<(PlayerId, MgmtPlayerInfo)>
215  {
216    let players = {
217      let MgmtGameResponseGameInfo { players, .. } = self.info()?;
218      players
219    };
220
221    players.into_iter().find(
222      |(_,mpi)| &mpi.account == account
223    )
224  }
225
226  #[throws(AE)]
227  pub fn list_pieces(&mut self) -> (Vec<MgmtGamePieceInfo>, BTreeSet<String>) {
228    let insns = vec![ MGI::ListPieces ];
229    let mut responses = self.alter_game(insns, None)?;
230    match responses.as_mut_slice() {
231      [MGR::Pieces { pieces, pcaliases }] => return (
232        mem::take(pieces),
233        mem::take(pcaliases),
234      ),
235      wat => Err(anyhow!("ListPieces => {:?}", &wat))?,
236    }
237  }
238
239  #[throws(AE)]
240  pub fn list_items(&mut self, lib: Option<String>, pat: String)
241                -> Vec<ItemEnquiryData> {
242    let cmd = MgmtCommand::LibraryListByGlob {
243      game: self.game.clone(),
244      lib, pat,
245    };
246    let mut items = match self.cmd(&cmd)? {
247      MgmtResponse::LibraryItems(items) => items,
248      wat => Err(anyhow!("unexpected LibraryListByGlob response: {:?}",
249                         &wat))?,
250    };
251    items.sort();
252    items
253  }
254
255/*
256  fn get_info(&mut self) -> Result<
257      (MgmtGameResponseGameInfo, HashMap<String,PlayerId>
258      ),AE>
259  {
260    let mut players = self.alter_game(
261      vec![ MgmtGameInstruction::Info ],
262      None,
263    )?;
264    let info = match players.pop() {
265      Some(MgmtGameResponse::Info(info)) => info,
266      wat => Err(anyhow!("GetGstate got {:?}", &wat))?,
267    };
268    let mut nick2id = HashMap::new();
269    for (player, pstate) in info.players.iter() {
270      use hash_map::Entry::*;
271      match nick2id.entry(pstate.nick.clone()) {
272        Occupied(oe) => Err(anyhow!("game has duplicate nick {:?}, {} {}",
273                                    &pstate.nick, *oe.get(), player))?,
274        Vacant(ve) => ve.insert(player),
275      };
276    }
277    Ok((info, nick2id))
278  }
279*/
280}
otter/mgmtchannel.rs

otter/
mgmtchannel.rs
