risedev/task/
pubsub_service.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::env;
16use std::path::{Path, PathBuf};
17use std::process::Command;
18
19use anyhow::{Result, anyhow};
20
21use super::{ExecuteContext, Task};
22use crate::util::stylized_risedev_subcmd;
23use crate::{DummyService, PubsubConfig};
24
25pub struct PubsubService {
26    config: PubsubConfig,
27}
28
29impl PubsubService {
30    pub fn new(config: PubsubConfig) -> Result<Self> {
31        Ok(Self { config })
32    }
33
34    fn gcloud_path(&self) -> Result<PathBuf> {
35        let prefix_bin = env::var("PREFIX_BIN")?;
36        Ok(Path::new(&prefix_bin)
37            .join("gcloud")
38            .join("start-pubsub-emulator.sh"))
39    }
40
41    fn gcloud(&self) -> Result<Command> {
42        Ok(Command::new(self.gcloud_path()?))
43    }
44}
45
46impl Task for PubsubService {
47    fn execute(&mut self, ctx: &mut ExecuteContext<impl std::io::Write>) -> anyhow::Result<()> {
48        if self.config.user_managed {
49            return DummyService::new(&self.id()).execute(ctx);
50        }
51
52        ctx.service(self);
53        ctx.pb.set_message("starting...");
54
55        let path = self.gcloud_path()?;
56        if !path.exists() {
57            return Err(anyhow!(
58                "gcloud binary not found in {:?}\nDid you enable pubsub-emulator feature in `{}`?",
59                path,
60                stylized_risedev_subcmd("configure")
61            ));
62        }
63
64        let mut cmd = self.gcloud()?;
65        cmd.arg(format!("{}", self.config.port));
66
67        ctx.run_command(ctx.tmux_run(cmd)?)?;
68
69        ctx.pb.set_message("started");
70
71        Ok(())
72    }
73
74    fn id(&self) -> String {
75        self.config.id.clone()
76    }
77}
risedev/task/pubsub_service.rs

risedev/task/
pubsub_service.rs
