diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml
index d8e31ed..f62d8a4 100644
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -45,7 +45,7 @@ jobs:
- name: Build
run: |
- bazel build --config=windows //cdc_rsync //cdc_stream //asset_stream_manager //tests_common //tests_asset_streaming_30 //tests_cdc_rsync
+ bazel build --config=windows //cdc_rsync //asset_stream_manager //tests_common //tests_asset_streaming_30 //tests_cdc_rsync
- name: Test
run: |
diff --git a/.github/workflows/create_release.yml b/.github/workflows/create_release.yml
index 51a827a..4d75d54 100644
--- a/.github/workflows/create_release.yml
+++ b/.github/workflows/create_release.yml
@@ -53,7 +53,7 @@ jobs:
- name: Build
run: |
bazel build --config=windows --compilation_mode=opt --copt=/GL `
- //cdc_rsync //cdc_stream //asset_stream_manager //tests_common //tests_asset_streaming_30 //tests_cdc_rsync
+ //cdc_rsync //asset_stream_manager //tests_common //tests_asset_streaming_30 //tests_cdc_rsync
- name: Test
run: |
@@ -77,7 +77,6 @@ jobs:
run: |
mkdir artifacts
cp bazel-bin/cdc_rsync/cdc_rsync.exe artifacts
- cp bazel-bin/cdc_stream/cdc_stream.exe artifacts
cp bazel-bin/asset_stream_manager/asset_stream_manager.exe artifacts
cp LICENSE artifacts
cp README.md artifacts
diff --git a/all_files.vcxitems b/all_files.vcxitems
index e28a753..3b5223b 100644
--- a/all_files.vcxitems
+++ b/all_files.vcxitems
@@ -21,6 +21,7 @@
+
@@ -30,7 +31,9 @@
+
+
@@ -42,8 +45,6 @@
-
-
@@ -150,6 +151,7 @@
+
@@ -157,7 +159,9 @@
+
+
@@ -166,7 +170,6 @@
-
@@ -246,7 +249,6 @@
-
diff --git a/asset_stream_manager/BUILD b/asset_stream_manager/BUILD
index c8f79b8..bc9c295 100644
--- a/asset_stream_manager/BUILD
+++ b/asset_stream_manager/BUILD
@@ -9,27 +9,71 @@ cc_binary(
srcs = ["main.cc"],
data = [":roots_pem"],
deps = [
- ":commands",
- "//cdc_stream",
+ ":start_command",
+ ":start_service_command",
+ ":stop_command",
"//common:log",
"//common:path",
],
)
cc_library(
- name = "commands",
- srcs = [
- "base_command.cc",
- "start_service_command.cc",
- ],
- hdrs = [
- "base_command.h",
- "start_service_command.h",
+ name = "base_command",
+ srcs = ["base_command.cc"],
+ hdrs = ["base_command.h"],
+ deps = [
+ "//absl_helper:jedec_size_flag",
+ "@com_github_lyra//:lyra",
+ "@com_google_absl//absl/status",
+ "@com_google_absl//absl/strings:str_format",
],
+)
+
+cc_library(
+ name = "start_service_command",
+ srcs = ["start_service_command.cc"],
+ hdrs = ["start_service_command.h"],
deps = [
":asset_stream_config",
+ ":base_command",
":session_management_server",
- "@com_github_lyra//:lyra",
+ ],
+)
+
+cc_library(
+ name = "start_command",
+ srcs = ["start_command.cc"],
+ hdrs = ["start_command.h"],
+ deps = [
+ ":base_command",
+ ":local_assets_stream_manager_client",
+ ":session_management_server",
+ "//common:path",
+ "//common:status_macros",
+ ],
+)
+
+cc_library(
+ name = "stop_command",
+ srcs = ["stop_command.cc"],
+ hdrs = ["stop_command.h"],
+ deps = [
+ ":base_command",
+ ":local_assets_stream_manager_client",
+ ":session_management_server",
+ "//common:path",
+ "//common:remote_util",
+ "//common:status_macros",
+ ],
+)
+
+cc_library(
+ name = "local_assets_stream_manager_client",
+ srcs = ["local_assets_stream_manager_client.cc"],
+ hdrs = ["local_assets_stream_manager_client.h"],
+ deps = [
+ "//common:grpc_status",
+ "//proto:local_assets_stream_manager_grpc_proto",
"@com_google_absl//absl/status",
],
)
@@ -66,8 +110,8 @@ cc_library(
srcs = ["asset_stream_config.cc"],
hdrs = ["asset_stream_config.h"],
deps = [
+ ":base_command",
":multi_session",
- "//absl_helper:jedec_size_flag",
"//common:log",
"//common:path",
"//common:status_macros",
diff --git a/asset_stream_manager/asset_stream_config.cc b/asset_stream_manager/asset_stream_config.cc
index db00123..3162713 100644
--- a/asset_stream_manager/asset_stream_config.cc
+++ b/asset_stream_manager/asset_stream_config.cc
@@ -19,6 +19,7 @@
#include "absl/strings/str_format.h"
#include "absl/strings/str_join.h"
#include "absl_helper/jedec_size_flag.h"
+#include "asset_stream_manager/base_command.h"
#include "common/buffer.h"
#include "common/path.h"
#include "common/status_macros.h"
@@ -38,12 +39,14 @@ AssetStreamConfig::AssetStreamConfig() = default;
AssetStreamConfig::~AssetStreamConfig() = default;
-void AssetStreamConfig::RegisterCommandLineFlags(lyra::command& cmd) {
+void AssetStreamConfig::RegisterCommandLineFlags(lyra::command& cmd,
+ BaseCommand& base_command) {
session_cfg_.verbosity = kDefaultVerbosity;
cmd.add_argument(lyra::opt(session_cfg_.verbosity, "num")
.name("--verbosity")
.help("Verbosity of the log output, default: " +
- std::to_string(kDefaultVerbosity)));
+ std::to_string(kDefaultVerbosity) +
+ ". Increase to make logs more verbose."));
cmd.add_argument(
lyra::opt(session_cfg_.stats)
@@ -84,13 +87,14 @@ void AssetStreamConfig::RegisterCommandLineFlags(lyra::command& cmd) {
.help("Check FUSE consistency and log check results"));
session_cfg_.fuse_cache_capacity = DiskDataStore::kDefaultCapacity;
- cmd.add_argument(lyra::opt(JedecParser("--cache-capacity",
+ cmd.add_argument(
+ lyra::opt(base_command.JedecParser("--cache-capacity",
&session_cfg_.fuse_cache_capacity),
- "bytes")
- .name("--cache-capacity")
- .help("FUSE cache capacity, default: " +
- std::to_string(DiskDataStore::kDefaultCapacity) +
- ". Supports common unit suffixes K, M, G."));
+ "bytes")
+ .name("--cache-capacity")
+ .help("FUSE cache capacity, default: " +
+ std::to_string(DiskDataStore::kDefaultCapacity) +
+ ". Supports common unit suffixes K, M, G."));
session_cfg_.fuse_cleanup_timeout_sec = DataProvider::kCleanupTimeoutSec;
cmd.add_argument(
@@ -251,18 +255,5 @@ std::string AssetStreamConfig::GetFlagReadErrors() {
error_str.empty() ? "" : "\n", flag, error);
return error_str;
}
-std::function AssetStreamConfig::JedecParser(
- const char* flag_name, uint64_t* bytes) {
- return [flag_name, bytes,
- error = &jedec_parse_error_](const std::string& value) {
- JedecSize size;
- if (AbslParseFlag(value, &size, error)) {
- *bytes = size.Size();
- } else {
- *error = absl::StrFormat("Failed to parse --%s=%s: %s", flag_name, value,
- *error);
- }
- };
-}
} // namespace cdc_ft
diff --git a/asset_stream_manager/asset_stream_config.h b/asset_stream_manager/asset_stream_config.h
index 08f6256..b240e51 100644
--- a/asset_stream_manager/asset_stream_config.h
+++ b/asset_stream_manager/asset_stream_config.h
@@ -31,6 +31,8 @@ class command;
namespace cdc_ft {
+class BaseCommand;
+
// Class containing all configuration settings for asset streaming.
// Reads flags from the command line and optionally applies overrides from
// a json file.
@@ -41,7 +43,7 @@ class AssetStreamConfig {
~AssetStreamConfig();
// Registers arguments with Lyra.
- void RegisterCommandLineFlags(lyra::command& cmd);
+ void RegisterCommandLineFlags(lyra::command& cmd, BaseCommand& base_command);
// Loads a configuration from the JSON file at |path| and overrides any config
// values that are set in this file. Sample json file:
@@ -88,16 +90,7 @@ class AssetStreamConfig {
// Whether to log to a file or to stdout.
bool log_to_stdout() const { return log_to_stdout_; }
- // Workaround for Lyra not accepting errors from parsers.
- const std::string& jedec_parse_error() const { return jedec_parse_error_; }
-
private:
- // Jedec parser for Lyra options. Usage:
- // lyra::opt(JedecParser("size-flag", &size_bytes), "bytes"))
- // Sets jedec_parse_error_ on error, Lyra doesn't support errors from lambdas.
- std::function JedecParser(const char* flag_name,
- uint64_t* bytes);
-
SessionConfig session_cfg_;
bool log_to_stdout_ = false;
@@ -111,9 +104,6 @@ class AssetStreamConfig {
// Maps flags to errors occurred while reading this flag.
std::map flag_read_errors_;
-
- // Errors from parsing JEDEC sizes.
- std::string jedec_parse_error_;
};
}; // namespace cdc_ft
diff --git a/asset_stream_manager/asset_stream_manager.vcxproj b/asset_stream_manager/asset_stream_manager.vcxproj
index 5189482..141ff00 100644
--- a/asset_stream_manager/asset_stream_manager.vcxproj
+++ b/asset_stream_manager/asset_stream_manager.vcxproj
@@ -66,7 +66,7 @@
- //asset_stream_manager //cdc_stream
+ //asset_stream_manager
asset_stream_manager.exe
..\;..\third_party\absl;..\bazel-cdc-file-transfer\external\com_github_jsoncpp\include;..\bazel-cdc-file-transfer\external\com_github_blake3\c;..\third_party\googletest\googletest\include;..\bazel-cdc-file-transfer\external\com_google_protobuf\src;..\bazel-cdc-file-transfer\external\com_github_grpc_grpc\include;..\bazel-out\x64_windows-dbg\bin;..\bazel-cdc-file-transfer\external\com_github_lyra\include;$(VC_IncludePath);$(WindowsSDK_IncludePath)
diff --git a/asset_stream_manager/base_command.cc b/asset_stream_manager/base_command.cc
index f211cf2..eaaf7ee 100644
--- a/asset_stream_manager/base_command.cc
+++ b/asset_stream_manager/base_command.cc
@@ -14,6 +14,8 @@
#include "asset_stream_manager/base_command.h"
+#include "absl/strings/str_format.h"
+#include "absl_helper/jedec_size_flag.h"
#include "lyra/lyra.hpp"
namespace cdc_ft {
@@ -33,14 +35,38 @@ void BaseCommand::Register(lyra::cli& cli) {
RegisterCommandLineFlags(cmd);
- // Workaround for Lyra treating --unknown_flags as positional argument.
- // If this argument is not empty, it's an unknown arg.
- cmd.add_argument(lyra::arg(invalid_arg_, ""));
+ // Detect extra positional args.
+ cmd.add_argument(lyra::arg(PosArgValidator(&extra_positional_arg_), ""));
// Register command with CLI.
cli.add_argument(std::move(cmd));
}
+std::function BaseCommand::JedecParser(
+ const char* flag_name, uint64_t* bytes) {
+ return [flag_name, bytes,
+ error = &jedec_parse_error_](const std::string& value) {
+ JedecSize size;
+ if (AbslParseFlag(value, &size, error)) {
+ *bytes = size.Size();
+ } else {
+ *error = absl::StrFormat("Failed to parse %s=%s: %s", flag_name, value,
+ *error);
+ }
+ };
+}
+
+std::function BaseCommand::PosArgValidator(
+ std::string* str) {
+ return [str, invalid_arg = &invalid_arg_](const std::string& value) {
+ if (!value.empty() && value[0] == '-') {
+ *invalid_arg = value;
+ } else {
+ *str = value;
+ }
+ };
+}
+
void BaseCommand::CommandHandler(const lyra::group& g) {
// Handle -h, --help.
if (show_help_) {
@@ -57,6 +83,19 @@ void BaseCommand::CommandHandler(const lyra::group& g) {
return;
}
+ if (!jedec_parse_error_.empty()) {
+ std::cerr << "Error: " << jedec_parse_error_ << std::endl;
+ *exit_code_ = 1;
+ return;
+ }
+
+ if (!extra_positional_arg_.empty()) {
+ std::cerr << "Error: Extraneous positional argument '"
+ << extra_positional_arg_ << "'. Try -h for help." << std::endl;
+ *exit_code_ = 1;
+ return;
+ }
+
// Run and print error.
absl::Status status = Run();
if (!status.ok()) {
diff --git a/asset_stream_manager/base_command.h b/asset_stream_manager/base_command.h
index 6c85bff..664448e 100644
--- a/asset_stream_manager/base_command.h
+++ b/asset_stream_manager/base_command.h
@@ -42,6 +42,18 @@ class BaseCommand {
// Registers the command with Lyra. Must be called before parsing args.
void Register(lyra::cli& cli);
+ // Jedec parser for Lyra options. Usage:
+ // lyra::opt(JedecParser("size-flag", &size_bytes), "bytes"))
+ // Automatically reports a parse failure on error.
+ std::function JedecParser(const char* flag_name,
+ uint64_t* bytes);
+
+ // Validator that should be used for all positional arguments. Lyra interprets
+ // -u, --unknown_flag as positional argument. This validator makes sure that
+ // a positional argument starting with - is reported as an error. Otherwise,
+ // writes the value to |str|.
+ std::function PosArgValidator(std::string* str);
+
protected:
// Adds all optional and required arguments used by the command.
// Called by Register().
@@ -66,6 +78,13 @@ class BaseCommand {
// Workaround for invalid args. Lyra doesn't interpret --invalid as invalid
// argument, but as positional argument "--invalid".
std::string invalid_arg_;
+
+ // Extraneous positional args. Gets reported as error if present.
+ std::string extra_positional_arg_;
+
+ // Errors from parsing JEDEC sizes.
+ // Works around Lyra not accepting errors from parsers.
+ std::string jedec_parse_error_;
};
} // namespace cdc_ft
diff --git a/cdc_stream/local_assets_stream_manager_client.cc b/asset_stream_manager/local_assets_stream_manager_client.cc
similarity index 64%
rename from cdc_stream/local_assets_stream_manager_client.cc
rename to asset_stream_manager/local_assets_stream_manager_client.cc
index 434e6d4..5503ea1 100644
--- a/cdc_stream/local_assets_stream_manager_client.cc
+++ b/asset_stream_manager/local_assets_stream_manager_client.cc
@@ -12,9 +12,13 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-#include "cdc_stream/local_assets_stream_manager_client.h"
+#include "asset_stream_manager/local_assets_stream_manager_client.h"
+
+#include
#include "absl/status/status.h"
+#include "absl/strings/str_format.h"
+#include "absl/strings/str_split.h"
#include "common/grpc_status.h"
namespace cdc_ft {
@@ -24,6 +28,15 @@ using StartSessionResponse = localassetsstreammanager::StartSessionResponse;
using StopSessionRequest = localassetsstreammanager::StopSessionRequest;
using StopSessionResponse = localassetsstreammanager::StopSessionResponse;
+LocalAssetsStreamManagerClient::LocalAssetsStreamManagerClient(
+ uint16_t service_port) {
+ std::string client_address = absl::StrFormat("localhost:%u", service_port);
+ std::shared_ptr channel = grpc::CreateCustomChannel(
+ client_address, grpc::InsecureChannelCredentials(),
+ grpc::ChannelArguments());
+ stub_ = LocalAssetsStreamManager::NewStub(std::move(channel));
+}
+
LocalAssetsStreamManagerClient::LocalAssetsStreamManagerClient(
std::shared_ptr channel) {
stub_ = LocalAssetsStreamManager::NewStub(std::move(channel));
@@ -59,4 +72,24 @@ absl::Status LocalAssetsStreamManagerClient::StopSession(
return ToAbslStatus(stub_->StopSession(&context, request, &response));
}
+// static
+absl::Status LocalAssetsStreamManagerClient::ParseUserHostDir(
+ const std::string& user_host_dir, std::string* user_host,
+ std::string* dir) {
+ std::vector parts =
+ absl::StrSplit(user_host_dir, absl::MaxSplits(':', 1));
+ if (parts.size() < 2 ||
+ (parts[0].size() == 1 && toupper(parts[0][0]) >= 'A' &&
+ toupper(parts[0][0]) <= 'Z')) {
+ return absl::InvalidArgumentError(
+ absl::StrFormat("Failed to parse '%s'. Make sure it is of the form "
+ "[user@]host:linux_dir.",
+ user_host_dir));
+ }
+
+ *user_host = parts[0];
+ *dir = parts[1];
+ return absl::OkStatus();
+}
+
} // namespace cdc_ft
diff --git a/cdc_stream/local_assets_stream_manager_client.h b/asset_stream_manager/local_assets_stream_manager_client.h
similarity index 64%
rename from cdc_stream/local_assets_stream_manager_client.h
rename to asset_stream_manager/local_assets_stream_manager_client.h
index 3aaf67e..cd535ca 100644
--- a/cdc_stream/local_assets_stream_manager_client.h
+++ b/asset_stream_manager/local_assets_stream_manager_client.h
@@ -32,26 +32,40 @@ namespace cdc_ft {
// gRpc client for starting/stopping asset streaming sessions.
class LocalAssetsStreamManagerClient {
public:
+ explicit LocalAssetsStreamManagerClient(uint16_t service_port);
+
// |channel| is a grpc channel to use.
explicit LocalAssetsStreamManagerClient(
std::shared_ptr channel);
+
~LocalAssetsStreamManagerClient();
- // Starts streaming the Windows directory |src_dir| to the Linux target
- // |user_host_dir|, which must be formatted as [user@]host:dir, e.g.
- // jdoe@jdoe.corp.foo.com:~/assets
+ // Starts streaming |src_dir| to |user_host|:|mount_dir|.
// Starting a second session to the same target will stop the first one.
+ // |src_dir| is the Windows source directory to stream.
+ // |user_host| is the Linux host, formatted as [user@:host].
+ // |ssh_port| is the SSH port to use while connecting to the host.
+ // |mount_dir| is the Linux target directory to stream to.
+ // |ssh_command| is the ssh command and extra arguments to use.
+ // |scp_command| is the scp command and extra arguments to use.
absl::Status StartSession(const std::string& src_dir,
const std::string& user_host, uint16_t ssh_port,
const std::string& mount_dir,
const std::string& ssh_command,
const std::string& scp_command);
- // Stops the streaming session to the Linux target |user_host_dir|, which must
- // be formatted as [user@]host:dir, e.g. jdoe@jdoe.corp.foo.com:~/assets
- absl::Status StopSession(const std::string& user_host_dir,
+ // Stops the streaming session to the Linux target |user_host|:|mount_dir|.
+ // |user_host| is the Linux host, formatted as [user@:host].
+ // |mount_dir| is the Linux target directory.
+ absl::Status StopSession(const std::string& user_host,
const std::string& mount_dir);
+ // Helper function that splits "user@host:dir" into "user@host" and "dir".
+ // Does not think that C: is a host.
+ static absl::Status ParseUserHostDir(const std::string& user_host_dir,
+ std::string* user_host,
+ std::string* dir);
+
private:
using LocalAssetsStreamManager =
localassetsstreammanager::LocalAssetsStreamManager;
diff --git a/asset_stream_manager/main.cc b/asset_stream_manager/main.cc
index 5cb2fbb..819b288 100644
--- a/asset_stream_manager/main.cc
+++ b/asset_stream_manager/main.cc
@@ -12,7 +12,9 @@
// See the License for the specific language governing permissions and
// limitations under the License.
+#include "asset_stream_manager/start_command.h"
#include "asset_stream_manager/start_service_command.h"
+#include "asset_stream_manager/stop_command.h"
#include "lyra/lyra.hpp"
int main(int argc, char* argv[]) {
@@ -22,8 +24,14 @@ int main(int argc, char* argv[]) {
int exit_code = -1;
cli.add_argument(lyra::help(show_help));
- cdc_ft::StartServiceCommand start_service(&exit_code);
- start_service.Register(cli);
+ cdc_ft::StartCommand start_cmd(&exit_code);
+ start_cmd.Register(cli);
+
+ cdc_ft::StopCommand stop_cmd(&exit_code);
+ stop_cmd.Register(cli);
+
+ cdc_ft::StartServiceCommand start_service_cmd(&exit_code);
+ start_service_cmd.Register(cli);
// Parse args and run. Note that parse actually runs the commands.
// exit_code is -1 if no command was run.
diff --git a/asset_stream_manager/session_management_server.h b/asset_stream_manager/session_management_server.h
index a1ff154..95bf76f 100644
--- a/asset_stream_manager/session_management_server.h
+++ b/asset_stream_manager/session_management_server.h
@@ -36,6 +36,8 @@ class ProcessFactory;
// - Background
class SessionManagementServer {
public:
+ static constexpr int kDefaultServicePort = 44432;
+
SessionManagementServer(grpc::Service* session_service,
grpc::Service* background_service,
SessionManager* session_manager);
diff --git a/asset_stream_manager/start_command.cc b/asset_stream_manager/start_command.cc
new file mode 100644
index 0000000..01894c1
--- /dev/null
+++ b/asset_stream_manager/start_command.cc
@@ -0,0 +1,110 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "asset_stream_manager/start_command.h"
+
+#include
+
+#include "asset_stream_manager/local_assets_stream_manager_client.h"
+#include "asset_stream_manager/session_management_server.h"
+#include "common/log.h"
+#include "common/path.h"
+#include "common/remote_util.h"
+#include "common/status_macros.h"
+#include "lyra/lyra.hpp"
+
+namespace cdc_ft {
+namespace {
+constexpr int kDefaultVerbosity = 2;
+} // namespace
+
+StartCommand::StartCommand(int* exit_code)
+ : BaseCommand("start",
+ "Start streaming files from a Windows to a Linux device",
+ exit_code) {}
+
+StartCommand::~StartCommand() = default;
+
+void StartCommand::RegisterCommandLineFlags(lyra::command& cmd) {
+ verbosity_ = kDefaultVerbosity;
+ cmd.add_argument(lyra::opt(verbosity_, "num")
+ .name("--verbosity")
+ .help("Verbosity of the log output, default: " +
+ std::to_string(kDefaultVerbosity) +
+ ". Increase to make logs more verbose."));
+
+ service_port_ = SessionManagementServer::kDefaultServicePort;
+ cmd.add_argument(
+ lyra::opt(service_port_, "port")
+ .name("--service-port")
+ .help("Local port to use while connecting to the local "
+ "asset stream service, default: " +
+ std::to_string(SessionManagementServer::kDefaultServicePort)));
+
+ ssh_port_ = RemoteUtil::kDefaultSshPort;
+ cmd.add_argument(
+ lyra::opt(ssh_port_, "port")
+ .name("--ssh-port")
+ .help("Port to use while connecting to the remote instance being "
+ "streamed to, default: " +
+ std::to_string(RemoteUtil::kDefaultSshPort)));
+
+ path::GetEnv("CDC_SSH_COMMAND", &ssh_command_).IgnoreError();
+ cmd.add_argument(
+ lyra::opt(ssh_command_, "ssh_command")
+ .name("--ssh-command")
+ .help("Path and arguments of ssh command to use, e.g. "
+ "\"C:\\path\\to\\ssh.exe -F config_file\". Can also be "
+ "specified by the CDC_SSH_COMMAND environment variable."));
+
+ path::GetEnv("CDC_SCP_COMMAND", &scp_command_).IgnoreError();
+ cmd.add_argument(
+ lyra::opt(scp_command_, "scp_command")
+ .name("--scp-command")
+ .help("Path and arguments of scp command to use, e.g. "
+ "\"C:\\path\\to\\scp.exe -F config_file\". Can also be "
+ "specified by the CDC_SCP_COMMAND environment variable."));
+
+ cmd.add_argument(lyra::arg(PosArgValidator(&src_dir_), "dir")
+ .required()
+ .help("Windows directory to stream"));
+
+ cmd.add_argument(
+ lyra::arg(PosArgValidator(&user_host_dir_), "[user@]host:src-dir")
+ .required()
+ .help("Linux host and directory to stream to"));
+}
+
+absl::Status StartCommand::Run() {
+ LogLevel level = Log::VerbosityToLogLevel(verbosity_);
+ ScopedLog scoped_log(std::make_unique(level));
+ LocalAssetsStreamManagerClient client(service_port_);
+
+ std::string full_src_dir = path::GetFullPath(src_dir_);
+ std::string user_host, mount_dir;
+ RETURN_IF_ERROR(LocalAssetsStreamManagerClient::ParseUserHostDir(
+ user_host_dir_, &user_host, &mount_dir));
+
+ absl::Status status =
+ client.StartSession(full_src_dir, user_host, ssh_port_, mount_dir,
+ ssh_command_, scp_command_);
+ if (status.ok()) {
+ LOG_INFO("Started streaming directory '%s' to '%s:%s'", src_dir_, user_host,
+ mount_dir);
+ }
+
+ return status;
+}
+
+} // namespace cdc_ft
diff --git a/asset_stream_manager/start_command.h b/asset_stream_manager/start_command.h
new file mode 100644
index 0000000..3fb8ef8
--- /dev/null
+++ b/asset_stream_manager/start_command.h
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef ASSET_STREAM_MANAGER_START_COMMAND_H_
+#define ASSET_STREAM_MANAGER_START_COMMAND_H_
+
+#include "absl/status/status.h"
+#include "asset_stream_manager/base_command.h"
+
+namespace cdc_ft {
+
+// Handler for the start command. Sends an RPC call to the service to starts a
+// new asset streaming session.
+class StartCommand : public BaseCommand {
+ public:
+ explicit StartCommand(int* exit_code);
+ ~StartCommand();
+
+ // BaseCommand:
+ void RegisterCommandLineFlags(lyra::command& cmd) override;
+ absl::Status Run() override;
+
+ private:
+ int verbosity_ = 0;
+ uint16_t service_port_ = 0;
+ uint16_t ssh_port_ = 0;
+ std::string ssh_command_;
+ std::string scp_command_;
+ std::string src_dir_;
+ std::string user_host_dir_;
+};
+
+} // namespace cdc_ft
+
+#endif // ASSET_STREAM_MANAGER_START_COMMAND_H_
diff --git a/asset_stream_manager/start_service_command.cc b/asset_stream_manager/start_service_command.cc
index 14ce97b..53d832f 100644
--- a/asset_stream_manager/start_service_command.cc
+++ b/asset_stream_manager/start_service_command.cc
@@ -30,8 +30,6 @@
namespace cdc_ft {
namespace {
-constexpr int kSessionManagementPort = 44432;
-
std::string GetLogPath(const char* log_dir, const char* log_base_name) {
DefaultSystemClock* clock = DefaultSystemClock::GetInstance();
std::string timestamp_ext = clock->FormatNow(".%Y%m%d-%H%M%S.log", false);
@@ -57,19 +55,17 @@ void StartServiceCommand::RegisterCommandLineFlags(lyra::command& cmd) {
.name("--log-dir")
.help("Directory to store log files, default: " + log_dir_));
- cfg_.RegisterCommandLineFlags(cmd);
+ cfg_.RegisterCommandLineFlags(cmd, *this);
}
absl::Status StartServiceCommand::Run() {
- if (!cfg_.jedec_parse_error().empty()) {
- return absl::InvalidArgumentError(cfg_.jedec_parse_error());
- }
-
// Set up config. Allow overriding this config with |config_file|.
absl::Status cfg_load_status = path::ExpandPathVariables(&config_file_);
cfg_load_status.Update(cfg_.LoadFromFile(config_file_));
- RETURN_IF_ERROR(InitLogging());
+ std::unique_ptr logger;
+ ASSIGN_OR_RETURN(logger, GetLogger());
+ cdc_ft::ScopedLog scoped_log(std::move(logger));
// Log status of loaded configuration. Errors are not critical.
if (cfg_load_status.ok()) {
@@ -102,16 +98,14 @@ absl::Status StartServiceCommand::Run() {
LOG_INFO("Asset stream manager shut down successfully.");
}
- Log::Shutdown();
return status;
}
-absl::Status StartServiceCommand::InitLogging() {
+absl::StatusOr> StartServiceCommand::GetLogger() {
LogLevel level = Log::VerbosityToLogLevel(cfg_.session_cfg().verbosity);
if (cfg_.log_to_stdout()) {
// Log to stdout.
- Log::Initialize(std::make_unique(level));
- return absl::OkStatus();
+ return std::make_unique(level);
}
// Log to file.
@@ -121,9 +115,8 @@ absl::Status StartServiceCommand::InitLogging() {
absl::StrFormat("Failed to create log directory '%s'", log_dir_));
}
- Log::Initialize(std::make_unique(
- level, GetLogPath(log_dir_.c_str(), "assets_stream_manager").c_str()));
- return absl::OkStatus();
+ return std::make_unique(
+ level, GetLogPath(log_dir_.c_str(), "assets_stream_manager").c_str());
}
// Runs the session management service and returns when it finishes.
@@ -154,7 +147,8 @@ absl::Status StartServiceCommand::RunService() {
RETURN_ABSL_IF_ERROR(
session_service.StartSession(nullptr, &request, &response));
}
- RETURN_IF_ERROR(sm_server.Start(kSessionManagementPort));
+ RETURN_IF_ERROR(
+ sm_server.Start(SessionManagementServer::kDefaultServicePort));
sm_server.RunUntilShutdown();
return absl::OkStatus();
}
diff --git a/asset_stream_manager/start_service_command.h b/asset_stream_manager/start_service_command.h
index 533dcb4..bcc1a59 100644
--- a/asset_stream_manager/start_service_command.h
+++ b/asset_stream_manager/start_service_command.h
@@ -17,6 +17,10 @@
#ifndef ASSET_STREAM_MANAGER_START_SERVICE_COMMAND_H_
#define ASSET_STREAM_MANAGER_START_SERVICE_COMMAND_H_
+#include
+
+#include "absl/status/status.h"
+#include "absl/status/statusor.h"
#include "asset_stream_manager/asset_stream_config.h"
#include "asset_stream_manager/base_command.h"
@@ -34,9 +38,8 @@ class StartServiceCommand : public BaseCommand {
absl::Status Run() override;
private:
- // Initializes LOG* logging.
- // Depending on the flags, might log to console or to a file.
- absl::Status InitLogging();
+ // Depending on the flags, returns a console or file logger.
+ absl::StatusOr> GetLogger();
// Runs the asset streaming service.
absl::Status RunService();
diff --git a/asset_stream_manager/stop_command.cc b/asset_stream_manager/stop_command.cc
new file mode 100644
index 0000000..5278efc
--- /dev/null
+++ b/asset_stream_manager/stop_command.cc
@@ -0,0 +1,75 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "asset_stream_manager/stop_command.h"
+
+#include
+
+#include "asset_stream_manager/local_assets_stream_manager_client.h"
+#include "asset_stream_manager/session_management_server.h"
+#include "common/log.h"
+#include "common/path.h"
+#include "common/status_macros.h"
+#include "lyra/lyra.hpp"
+
+namespace cdc_ft {
+namespace {
+constexpr int kDefaultVerbosity = 2;
+} // namespace
+
+StopCommand::StopCommand(int* exit_code)
+ : BaseCommand("stop", "Stops a streaming session", exit_code) {}
+
+StopCommand::~StopCommand() = default;
+
+void StopCommand::RegisterCommandLineFlags(lyra::command& cmd) {
+ verbosity_ = kDefaultVerbosity;
+ cmd.add_argument(lyra::opt(verbosity_, "num")
+ .name("--verbosity")
+ .help("Verbosity of the log output, default: " +
+ std::to_string(kDefaultVerbosity) +
+ ". Increase to make logs more verbose."));
+
+ service_port_ = SessionManagementServer::kDefaultServicePort;
+ cmd.add_argument(
+ lyra::opt(service_port_, "port")
+ .name("--service-port")
+ .help("Local port to use while connecting to the local "
+ "asset stream service, default: " +
+ std::to_string(SessionManagementServer::kDefaultServicePort)));
+
+ cmd.add_argument(
+ lyra::arg(PosArgValidator(&user_host_dir_), "[user@]host:src-dir")
+ .required()
+ .help("Linux host and directory to stream to"));
+}
+
+absl::Status StopCommand::Run() {
+ LogLevel level = Log::VerbosityToLogLevel(verbosity_);
+ ScopedLog scoped_log(std::make_unique(level));
+ LocalAssetsStreamManagerClient client(service_port_);
+
+ std::string user_host, mount_dir;
+ RETURN_IF_ERROR(LocalAssetsStreamManagerClient::ParseUserHostDir(
+ user_host_dir_, &user_host, &mount_dir));
+
+ absl::Status status = client.StopSession(user_host, mount_dir);
+ if (status.ok()) {
+ LOG_INFO("Stopped streaming session to '%s:%s'", user_host, mount_dir);
+ }
+
+ return status;
+}
+
+} // namespace cdc_ft
diff --git a/asset_stream_manager/stop_command.h b/asset_stream_manager/stop_command.h
new file mode 100644
index 0000000..12b35ac
--- /dev/null
+++ b/asset_stream_manager/stop_command.h
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef ASSET_STREAM_MANAGER_STOP_COMMAND_H_
+#define ASSET_STREAM_MANAGER_STOP_COMMAND_H_
+
+#include "absl/status/status.h"
+#include "asset_stream_manager/base_command.h"
+
+namespace cdc_ft {
+
+// Handler for the stop command. Sends an RPC call to the service to stop an
+// asset streaming session.
+class StopCommand : public BaseCommand {
+ public:
+ explicit StopCommand(int* exit_code);
+ ~StopCommand();
+
+ // BaseCommand:
+ void RegisterCommandLineFlags(lyra::command& cmd) override;
+ absl::Status Run() override;
+
+ private:
+ int verbosity_ = 0;
+ uint16_t service_port_ = 0;
+ std::string user_host_dir_;
+};
+
+} // namespace cdc_ft
+
+#endif // ASSET_STREAM_MANAGER_STOP_COMMAND_H_
diff --git a/cdc_stream/.gitignore b/cdc_stream/.gitignore
deleted file mode 100644
index 7dc8dde..0000000
--- a/cdc_stream/.gitignore
+++ /dev/null
@@ -1,3 +0,0 @@
-x64/*
-*.log
-*.user
\ No newline at end of file
diff --git a/cdc_stream/BUILD b/cdc_stream/BUILD
deleted file mode 100644
index 5b6fe22..0000000
--- a/cdc_stream/BUILD
+++ /dev/null
@@ -1,28 +0,0 @@
-load("@rules_cc//cc:defs.bzl", "cc_binary", "cc_library")
-
-package(default_visibility = [
- "//:__subpackages__",
-])
-
-cc_binary(
- name = "cdc_stream",
- srcs = ["main.cc"],
- deps = [
- ":local_assets_stream_manager_client",
- "//common:log",
- "//common:path",
- "@com_google_absl//absl/flags:parse",
- "@com_google_absl//absl/status",
- ],
-)
-
-cc_library(
- name = "local_assets_stream_manager_client",
- srcs = ["local_assets_stream_manager_client.cc"],
- hdrs = ["local_assets_stream_manager_client.h"],
- deps = [
- "//common:grpc_status",
- "//proto:local_assets_stream_manager_grpc_proto",
- "@com_google_absl//absl/status",
- ],
-)
diff --git a/cdc_stream/main.cc b/cdc_stream/main.cc
deleted file mode 100644
index 7fa5891..0000000
--- a/cdc_stream/main.cc
+++ /dev/null
@@ -1,165 +0,0 @@
-// Copyright 2022 Google LLC
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-#include
-#include
-
-#include "absl/flags/flag.h"
-#include "absl/flags/parse.h"
-#include "absl/flags/usage.h"
-#include "absl/status/status.h"
-#include "absl/strings/str_split.h"
-#include "cdc_stream/local_assets_stream_manager_client.h"
-#include "common/log.h"
-#include "common/path.h"
-#include "grpcpp/channel.h"
-#include "grpcpp/security/credentials.h"
-
-ABSL_FLAG(uint16_t, service_port, 44432,
- "Port to use while connecting to the local asset stream service.");
-ABSL_FLAG(
- uint16_t, ssh_port, 22,
- "Port to use while connecting to the remote instance being streamed to.");
-ABSL_FLAG(std::string, ssh_command, "",
- "Path and arguments of ssh command to use, e.g. "
- "\"C:\\path\\to\\ssh.exe -F config_file\". Can also be specified by "
- "the CDC_SSH_COMMAND environment variable.");
-ABSL_FLAG(std::string, scp_command, "",
- "Path and arguments of scp command to use, e.g. "
- "\"C:\\path\\to\\scp.exe -F config_file\". Can also be specified by "
- "the CDC_SCP_COMMAND environment variable.");
-ABSL_FLAG(int, verbosity, 2,
- "Verbosity of the log output. Increase to make logs more verbose.");
-
-namespace {
-
-constexpr char kHelpText[] = R"!(Stream files from a Windows to a Linux device
-
-Usage: cdc_stream [flags] start windows_dir [user@]host:linux_dir
- Streams the Windows directory windows_dir to directory linux_dir on the
- Linux host using SSH username user.
-
- cdc_stream [flags] stop [user@]host:linux_dir
- Stops the streaming session to the given Linux target.
-
-Type cdc_stream --helpfull for available flags.)!";
-
-// Splits |user_host_dir| = [user@]host:dir up into [user@]host and dir.
-// Does not touch Windows drives, e.g. C:\foo.
-bool ParseUserHostDir(const std::string& user_host_dir, std::string* user_host,
- std::string* dir) {
- std::vector parts =
- absl::StrSplit(user_host_dir, absl::MaxSplits(':', 1));
- if (parts.size() < 2 ||
- (parts[0].size() == 1 && toupper(parts[0][0]) >= 'A' &&
- toupper(parts[0][0]) <= 'Z')) {
- LOG_ERROR(
- "Failed to parse '%s'. Make sure it is of the form "
- "[user@]host:linux_dir.",
- user_host_dir);
- return false;
- }
-
- *user_host = parts[0];
- *dir = parts[1];
- return true;
-}
-
-std::string GetFlagFromEnvOrArgs(const char* env,
- const absl::Flag& flag) {
- std::string value = absl::GetFlag(flag);
- if (value.empty()) {
- cdc_ft::path::GetEnv(env, &value).IgnoreError();
- }
- return value;
-}
-
-} // namespace
-
-int main(int argc, char* argv[]) {
- absl::SetProgramUsageMessage(kHelpText);
- std::vector args = absl::ParseCommandLine(argc, argv);
-
- uint16_t service_port = absl::GetFlag(FLAGS_service_port);
- int verbosity = absl::GetFlag(FLAGS_verbosity);
- std::string ssh_command =
- GetFlagFromEnvOrArgs("CDC_SSH_COMMAND", FLAGS_ssh_command);
- std::string scp_command =
- GetFlagFromEnvOrArgs("CDC_SCP_COMMAND", FLAGS_scp_command);
- uint16_t ssh_port = absl::GetFlag(FLAGS_ssh_port);
-
- cdc_ft::LogLevel level = cdc_ft::Log::VerbosityToLogLevel(verbosity);
- cdc_ft::Log::Initialize(std::make_unique(level));
-
- if (args.size() < 2) {
- LOG_INFO(kHelpText);
- return 0;
- }
- std::string command = args[1];
- if (command != "start" && command != "stop") {
- LOG_ERROR("Unknown command '%s'. Must be 'start' or 'stop'.", command);
- return 1;
- }
-
- // Start a gRpc client.
- std::string client_address = absl::StrFormat("localhost:%u", service_port);
- std::shared_ptr grpc_channel = grpc::CreateCustomChannel(
- client_address, grpc::InsecureChannelCredentials(),
- grpc::ChannelArguments());
-
- cdc_ft::LocalAssetsStreamManagerClient client(grpc_channel);
-
- absl::Status status;
- if (command == "start") {
- if (args.size() < 4) {
- LOG_ERROR(
- "Command 'start' needs 2 arguments: the Windows directory to stream"
- " and the Linux target [user@]host:dir.");
- return 1;
- }
-
- std::string src_dir = cdc_ft::path::GetFullPath(args[2]);
- std::string user_host, mount_dir;
- if (!ParseUserHostDir(args[3], &user_host, &mount_dir)) return 1;
-
- status = client.StartSession(src_dir, user_host, ssh_port, mount_dir,
- ssh_command, scp_command);
- if (status.ok()) {
- LOG_INFO("Started streaming directory '%s' to '%s:%s'", src_dir,
- user_host, mount_dir);
- }
- } else /* if (command == "stop") */ {
- if (args.size() < 3) {
- LOG_ERROR(
- "Command 'stop' needs 1 argument: the Linux target "
- "[user@]host:linux_dir.");
- return 1;
- }
-
- std::string user_host, mount_dir;
- if (!ParseUserHostDir(args[2], &user_host, &mount_dir)) return 1;
-
- status = client.StopSession(user_host, mount_dir);
- if (status.ok()) {
- LOG_INFO("Stopped streaming session to '%s:%s'", user_host, mount_dir);
- }
- }
-
- if (!status.ok()) {
- LOG_ERROR("Error: %s", status.ToString());
- }
-
- cdc_ft::Log::Shutdown();
- return static_cast(status.code());
-}
diff --git a/common/log.h b/common/log.h
index 4f15e04..fc681df 100644
--- a/common/log.h
+++ b/common/log.h
@@ -123,6 +123,13 @@ class ConsoleLog : public Log {
absl::Mutex mutex_;
};
+// Initializes the log in the constructor and shuts it down on destruction.
+class ScopedLog {
+ public:
+ ScopedLog(std::unique_ptr log) { Log::Initialize(std::move(log)); }
+ ~ScopedLog() { Log::Shutdown(); }
+};
+
class FileLog : public Log {
public:
FileLog(LogLevel log_level, const char* path);