action.h

Go to the documentation of this file.

// Copyright 2025 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 ACTIONENGINE_ACTIONS_ACTION_H_
#define ACTIONENGINE_ACTIONS_ACTION_H_
 
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
 
#include <absl/base/nullability.h>
#include <absl/base/thread_annotations.h>
#include <absl/container/flat_hash_map.h>
#include <absl/container/flat_hash_set.h>
#include <absl/status/status.h>
#include <absl/time/time.h>
 
#include "actionengine/actions/registry.h"
#include "actionengine/actions/schema.h"
#include "actionengine/concurrency/concurrency.h"
#include "actionengine/data/types.h"
#include "actionengine/net/stream.h"
#include "actionengine/nodes/async_node.h"
#include "actionengine/nodes/node_map.h"
#include "actionengine/stores/chunk_store_reader.h"
 
namespace act {
class Session;
}  // namespace act

 
namespace act {

class Action : public std::enable_shared_from_this<Action> {
 public:
  explicit Action(ActionSchema schema, std::string_view id = "",
                  std::vector<Port> inputs = {},
                  std::vector<Port> outputs = {});
 
  ~Action();

  [[nodiscard]] ActionMessage GetActionMessage() const;

  AsyncNode* absl_nullable GetNode(std::string_view id);

  AsyncNode* absl_nullable GetInput(
      std::string_view name, std::optional<bool> bind_stream = std::nullopt);

  AsyncNode* absl_nullable GetOutput(
      std::string_view name,
      const std::optional<bool> bind_stream = std::nullopt) {
    act::MutexLock lock(&mu_);
    return GetOutputInternal(name, bind_stream);
  }

  void BindHandler(ActionHandler handler) { handler_ = std::move(handler); }
 
  void BindNodeMap(NodeMap* absl_nullable node_map);

  [[nodiscard]] NodeMap* absl_nullable GetNodeMap() const;

  void BindStream(WireStream* absl_nullable stream);

  [[nodiscard]] WireStream* absl_nullable GetStream() const;

  void BindSession(Session* absl_nullable session);

  [[nodiscard]] Session* absl_nullable GetSession() const;

  std::unique_ptr<Action> MakeActionInSameSession(
      std::string_view name, std::string_view action_id = "") const;

  [[nodiscard]] ActionRegistry* absl_nullable GetRegistry() const;

  [[nodiscard]] std::string GetId() const { return id_; }
 
  [[nodiscard]] const ActionSchema& GetSchema() const { return schema_; }

  absl::Status Await(absl::Duration timeout = absl::InfiniteDuration());

  absl::Status Call();

  absl::Status Run();

  void ClearInputsAfterRun(bool clear = true);

  void ClearOutputsAfterRun(bool clear = true);

  void Cancel() const {
    act::MutexLock lock(&mu_);
    CancelInternal();
  }

  thread::Case OnCancel() const { return cancelled_->OnEvent(); }

  [[nodiscard]] bool Cancelled() const;
 
  void BindStreamsOnInputsByDefault(bool bind) {
    act::MutexLock lock(&mu_);
    bind_streams_on_inputs_default_ = bind;
  }
 
  void BindStreamsOnOutputsByDefault(bool bind) {
    act::MutexLock lock(&mu_);
    bind_streams_on_outputs_default_ = bind;
  }
 
  void SetUserData(std::shared_ptr<void> data) {
    act::MutexLock lock(&mu_);
    user_data_ = std::move(data);
  }
 
  [[nodiscard]] void* absl_nullable GetUserData() const {
    act::MutexLock lock(&mu_);
    return user_data_.get();
  }
 
 private:
  void CancelInternal() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_);
 
  // Implementation detail: gets the input node ID for the given name, unique
  // to this particular action run/call.
  std::string GetInputId(std::string_view name) const;
 
  // Implementation detail: gets the input node ID for the given name, unique
  // to this particular action run/call.
  std::string GetOutputId(std::string_view name) const;
 
  AsyncNode* absl_nonnull GetOutputInternal(
      std::string_view name, std::optional<bool> bind_stream = std::nullopt)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_);
 
  void UnbindStreams() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_);
 
  mutable act::Mutex mu_{};
  act::CondVar cv_ ABSL_GUARDED_BY(mu_);
 
  ActionSchema schema_;
  absl::flat_hash_map<std::string, std::string> input_name_to_id_;
  absl::flat_hash_map<std::string, std::string> output_name_to_id_;
 
  ActionHandler handler_;
  bool has_been_called_ ABSL_GUARDED_BY(mu_) = false;
  bool has_been_run_ ABSL_GUARDED_BY(mu_) = false;
  std::string id_;
 
  NodeMap* absl_nullable node_map_ ABSL_GUARDED_BY(mu_) = nullptr;
  WireStream* absl_nullable stream_ ABSL_GUARDED_BY(mu_) = nullptr;
  Session* absl_nullable session_ ABSL_GUARDED_BY(mu_) = nullptr;
 
  absl::flat_hash_set<ChunkStoreReader*> reffed_readers_ ABSL_GUARDED_BY(mu_);
 
  bool bind_streams_on_inputs_default_ = true;
  bool bind_streams_on_outputs_default_ = false;
  absl::flat_hash_set<AsyncNode*> nodes_with_bound_streams_
      ABSL_GUARDED_BY(mu_);
 
  std::unique_ptr<thread::PermanentEvent> cancelled_;
 
  bool clear_inputs_after_run_ ABSL_GUARDED_BY(mu_) = false;
  bool clear_outputs_after_run_ ABSL_GUARDED_BY(mu_) = false;
  std::optional<absl::Status> run_status_ ABSL_GUARDED_BY(mu_) = std::nullopt;
 
  std::shared_ptr<void> user_data_ ABSL_GUARDED_BY(mu_) = nullptr;
};
 
}  // namespace act
 
#endif  // ACTIONENGINE_ACTIONS_ACTION_H_