Provider - check debit notes acceptance

agent/provider/Cargo.toml

@@ -16,8 +16,8 @@ ya-utils-actix = "0.1"
 ya-utils-path = "0.1"
 ya-utils-process = { version = "0.1", features = ['lock'] }
 
-actix = { version = "0.9", default-features = false }
-actix-rt = "1.0"
+actix = { version = "0.10", default-features = false }
+actix-rt = "1.1.1"
 actix_derive = "0.5.0"
 anyhow = "1.0"
 backoff = "0.2.1"

agent/provider/src/display.rs

@@ -0,0 +1,24 @@
+use humantime::format_duration;
+use std::fmt::{Error, Formatter};
+
+pub struct DisplayEnabler<'a, Type>(pub &'a Type);
+
+pub trait EnableDisplay<Type> {
+    fn display(&self) -> DisplayEnabler<Type>;
+}
+
+impl<Type> EnableDisplay<Type> for Type {
+    fn display(&self) -> DisplayEnabler<Type> {
+        DisplayEnabler(self)
+    }
+}
+
+impl<'a> std::fmt::Display for DisplayEnabler<'a, chrono::Duration> {
+    fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error> {
+        match self.0.clone().to_std() {
+            Ok(duration) => write!(f, "{}", format_duration(duration)),
+            // If we can't convert, we will display ugly version, which we wanted to avoid.
+            Err(_) => write!(f, "{}", self.0),
+        }
+    }
+}

agent/provider/src/execution/task_runner.rs

@@ -21,7 +21,7 @@ use ya_client_model::activity::{ActivityState, ProviderEvent, State};
 use ya_core_model::activity;
 use ya_utils_actix::actix_handler::ResultTypeGetter;
 use ya_utils_actix::actix_signal::{SignalSlot, Subscribe};
-use ya_utils_actix::forward_actix_handler;
+use ya_utils_actix::{actix_signal_handler, forward_actix_handler};
 use ya_utils_path::SecurePath;
 use ya_utils_process::ExeUnitExitStatus;
 
@@ -364,7 +364,7 @@ impl TaskRunner {
         _ctx: &mut Context<Self>,
     ) -> Result<()> {
         // Agreement waits for first create activity event.
-        let agreement_id = msg.agreement.agreement_id.clone();
+        let agreement_id = msg.agreement.id.clone();
         self.active_agreements.insert(agreement_id, msg.agreement);
         Ok(())
     }
@@ -462,22 +462,6 @@ impl TaskRunner {
         Ok(())
     }
 
-    pub fn on_subscribe_activity_created(
-        &mut self,
-        msg: Subscribe<ActivityCreated>,
-        _ctx: &mut Context<Self>,
-    ) -> Result<()> {
-        Ok(self.activity_created.on_subscribe(msg))
-    }
-
-    pub fn on_subscribe_activity_destroyed(
-        &mut self,
-        msg: Subscribe<ActivityDestroyed>,
-        _ctx: &mut Context<Self>,
-    ) -> Result<()> {
-        Ok(self.activity_destroyed.on_subscribe(msg))
-    }
-
     fn list_activities(&self, agreement_id: &str) -> Vec<String> {
         self.tasks
             .iter()
@@ -557,6 +541,8 @@ impl Actor for TaskRunner {
 forward_actix_handler!(TaskRunner, AgreementApproved, on_agreement_approved);
 forward_actix_handler!(TaskRunner, ExeUnitProcessFinished, on_exeunit_exited);
 forward_actix_handler!(TaskRunner, GetExeUnit, get_exeunit);
+actix_signal_handler!(TaskRunner, ActivityCreated, activity_created);
+actix_signal_handler!(TaskRunner, ActivityDestroyed, activity_destroyed);
 
 impl Handler<GetOfferTemplates> for TaskRunner {
     type Result = ResponseFuture<Result<HashMap<String, OfferTemplate>>>;
@@ -582,17 +568,6 @@ impl Handler<GetOfferTemplates> for TaskRunner {
     }
 }
 
-forward_actix_handler!(
-    TaskRunner,
-    Subscribe<ActivityCreated>,
-    on_subscribe_activity_created
-);
-forward_actix_handler!(
-    TaskRunner,
-    Subscribe<ActivityDestroyed>,
-    on_subscribe_activity_destroyed
-);
-
 impl Handler<UpdateActivity> for TaskRunner {
     type Result = ActorResponse<Self, (), Error>;
 

agent/provider/src/main.rs

@@ -4,6 +4,7 @@ use structopt::{clap, StructOpt};
 
 mod cli;
 mod dir;
+mod display;
 mod events;
 mod execution;
 mod hardware;

agent/provider/src/market/config.rs

@@ -1,5 +1,7 @@
 use structopt::StructOpt;
 
+use crate::market::negotiator::factory::NegotiatorsConfig;
+
 /// Configuration for ProviderMarket actor.
 #[derive(StructOpt, Clone, Debug)]
 pub struct MarketConfig {
@@ -9,8 +11,10 @@ pub struct MarketConfig {
     pub negotiation_events_interval: f32,
     #[structopt(long, env, default_value = "10.0")]
     pub agreement_approve_timeout: f32,
-    #[structopt(long, env, default_value = "LimitAgreements")]
+    #[structopt(long, env, default_value = "Composite")]
     pub negotiator_type: String,
+    #[structopt(flatten)]
+    pub negotiator_config: NegotiatorsConfig,
     #[structopt(skip = "you-forgot-to-set-session-id")]
     pub session_id: String,
 }

agent/provider/src/market/mod.rs

@@ -1,5 +1,4 @@
 pub mod config;
-mod mock_negotiator;
 mod negotiator;
 pub mod presets;
 pub mod provider_market;

agent/provider/src/market/negotiator.rs

@@ -1,99 +1,17 @@
-use ya_agreement_utils::AgreementView;
-use ya_agreement_utils::OfferDefinition;
-use ya_client_model::market::{NewOffer, Proposal};
-
-use anyhow::Result;
-use derive_more::Display;
-
-use crate::market::termination_reason::BreakReason;
-use ya_client::model::market::Reason;
-
-/// Response for requestor proposals.
-#[derive(Debug, Display)]
-#[allow(dead_code)]
-pub enum ProposalResponse {
-    #[display(fmt = "CounterProposal")]
-    CounterProposal {
-        offer: NewOffer,
-    },
-    AcceptProposal,
-    #[display(
-        fmt = "RejectProposal{}",
-        "reason.as_ref().map(|r| format!(\" (reason: {})\", r)).unwrap_or(\"\".into())"
-    )]
-    RejectProposal {
-        reason: Option<Reason>,
-    },
-    ///< Don't send any message to requestor. Could be useful to wait for other offers.
-    IgnoreProposal,
-}
-
-/// Response for requestor agreements.
-#[derive(Debug, Display)]
-#[allow(dead_code)]
-pub enum AgreementResponse {
-    ApproveAgreement,
-    #[display(
-        fmt = "RejectAgreement{}",
-        "reason.as_ref().map(|r| format!(\" (reason: {})\", r)).unwrap_or(\"\".into())"
-    )]
-    RejectAgreement {
-        reason: Option<Reason>,
-    },
-}
-
-/// Result of agreement execution.
-pub enum AgreementResult {
-    /// Failed to approve agreement. (Agreement even wasn't created)
-    ApprovalFailed,
-    /// Agreement was finished with success after first Activity.
-    ClosedByUs,
-    /// Agreement was finished with success by Requestor.
-    ClosedByRequestor,
-    /// Agreement was broken by us.
-    Broken { reason: BreakReason },
-}
-
-pub trait Negotiator {
-    /// Negotiator can modify offer, that was generated for him. He can save
-    /// information about this offer, that are necessary for negotiations.
-    fn create_offer(&mut self, node_info: &OfferDefinition) -> Result<NewOffer>;
-
-    /// Agreement notifications. Negotiator can adjust his strategy based on it.
-    fn agreement_finalized(&mut self, agreement_id: &str, result: &AgreementResult) -> Result<()>;
-
-    /// Reactions to events from market. These function make market decisions.
-    fn react_to_proposal(
-        &mut self,
-        offer: &NewOffer,
-        demand: &Proposal,
-    ) -> Result<ProposalResponse>;
-    fn react_to_agreement(&mut self, agreement: &AgreementView) -> Result<AgreementResponse>;
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_proposal_response_display() {
-        let reason = ProposalResponse::RejectProposal {
-            reason: Some("zima".into()),
-        };
-        let no_reason = ProposalResponse::RejectProposal { reason: None };
-
-        assert_eq!(reason.to_string(), "RejectProposal (reason: 'zima')");
-        assert_eq!(no_reason.to_string(), "RejectProposal");
-    }
-
-    #[test]
-    fn test_agreement_response_display() {
-        let reason = AgreementResponse::RejectAgreement {
-            reason: Some("lato".into()),
-        };
-        let no_reason = AgreementResponse::RejectAgreement { reason: None };
-
-        assert_eq!(reason.to_string(), "RejectAgreement (reason: 'lato')");
-        assert_eq!(no_reason.to_string(), "RejectAgreement");
-    }
-}
+mod accept_all;
+mod builtin;
+mod common;
+mod component;
+mod composite;
+pub mod factory;
+mod limit_agreements;
+
+pub use accept_all::AcceptAllNegotiator;
+pub use composite::CompositeNegotiator;
+pub use limit_agreements::LimitAgreementsNegotiator;
+
+pub use common::{
+    AgreementResponse, AgreementResult, Negotiator, NegotiatorAddr, ProposalResponse,
+};
+
+pub use component::{NegotiationResult, NegotiatorComponent, NegotiatorsPack, ProposalView};