dunedaq::snbmodules::TransferClient Class Reference

TransferClient class, represent a client that can create session and communicate with Bookkeepers. extend NotificationInterface because the client can communicate and need to initialize the connection. More...

#include <transfer_client.hpp>

Inheritance diagram for dunedaq::snbmodules::TransferClient:

Collaboration diagram for dunedaq::snbmodules::TransferClient:

Public Member Functions
	TransferClient (const IPFormat &listening_ip, const std::string &client_id, const std::filesystem::path &listening_dir, const std::string &connection_prefix="snbmodules", int timeout_send=10, int timeout_receive=100)
	TransferClient constructor.
	~TransferClient ()
bool	start (int timeout)
	Start the client in curent thread.
bool	do_work (std::atomic< bool > &running_flag)
	Start function to use in a thread.
void	create_new_transfer (const std::string &transfer_id, const std::string &protocol, const std::set< std::string > &dest_clients, const std::set< std::filesystem::path > &files, const nlohmann::json &protocol_options=nlohmann::json())
	Create a new transfer.
void	start_transfer (const std::string &transfer_id)
	Start, pause, resume or cancel a transfer.
void	pause_transfer (const std::string &transfer_id)
void	resume_transfer (const std::string &transfer_id)
void	cancel_transfer (const std::string &transfer_id)
TransferSession &	create_session (GroupMetadata transfer_options, e_session_type type, std::string id, const std::filesystem::path &work_dir, IPFormat ip=IPFormat(), const std::set< std::string > &dest_clients=std::set< std::string >())
	Create a new session, you can precise the IP address of the session forcing to use a specific network interface and different port that the client.
void	scan_available_files (std::set< std::filesystem::path > &previous_scan, bool nested=false, std::filesystem::path folder=std::filesystem::path())
	Scan available files in the listening directory.
IPFormat	get_ip () const
std::string	get_client_id () const
std::filesystem::path	get_listening_dir () const
TransferSession *	get_session (std::string transfer_id)
std::vector< TransferSession > &	get_sessions ()
const std::vector< TransferSession > &	get_sessions () const
std::string	get_my_conn ()
void	set_ip (const std::string &ip)
void	set_port (int port)
void	set_client_id (std::string client_id)
void	set_listening_dir (const std::filesystem::path &listening_dir)
Public Member Functions inherited from dunedaq::snbmodules::NotificationInterface
	NotificationInterface (std::string connection_prefix="snbmodules", int timeout_send=10, int timeout_receive=100)
	NotificationInterface (std::vector< std::string > bk_conn, std::set< std::string > client_conn, const std::string &connection_prefix="snbmodules", int timeout_send=10, int timeout_receive=100)
virtual	~NotificationInterface ()=default
std::optional< NotificationData >	listen_for_notification (const std::string &id, const std::string &expected_from="", int timeout=-1, int tries=-1)
	Listen for a notification.
bool	send_notification (const notification_type::e_notification_type &notif, const std::string &src, const std::string &dst, const std::string &id_conn, const std::string &data="", int tries=-1)
	Send a notification during m_timeout_send ms.
void	lookups_connections ()
	Get the list of every connections, must have the prefix first in the name of the connection and the name of the connection must be in the format: prefix.*bookkeeper.* or prefix.*client.*.
const std::vector< std::string > &	get_bookkeepers_conn () const
	Init the connection interface, Only used for standalone application.
const std::set< std::string > &	get_clients_conn () const

Private Member Functions
std::string	generate_session_id (const std::string &transferid, const std::string &dest="")
	Function to generate session ID.
void	remove_session (const std::string &session_id)
	Add a session to the client.
bool	action_on_receive_notification (NotificationData notif) override
	Action to do when the client receive a notification.
void	share_available_files (const std::set< std::filesystem::path > &to_share, const std::string &dest)
	Share available files (in m_listening_dir)
std::shared_ptr< TransferMetadata >	create_metadata_from_file (const std::filesystem::path &src)
	Create a metadata from a file.

Private Attributes
IPFormat	m_listening_ip
	IP address of the client.
std::string	m_client_id
	TransferClient ID, unique identifier of the client.
std::filesystem::path	m_listening_dir
	Listening directory, directory where the client will listen for incoming files and files to share.
std::vector< TransferSession >	m_sessions
	Map of active sessions (key = session ID, value = session)
std::map< std::string, std::shared_ptr< TransferMetadata > >	m_available_files
	Map of available files (key = file path, value = file metadata)
std::string	m_my_conn = ""
	Connection uuid of the client, retrieved using the notification interface and calling get_my_conn()

Detailed Description

TransferClient class, represent a client that can create session and communicate with Bookkeepers. extend NotificationInterface because the client can communicate and need to initialize the connection.

Definition at line 33 of file transfer_client.hpp.

Constructor & Destructor Documentation

TransferClient()

dunedaq::snbmodules::TransferClient::TransferClient	(	const IPFormat &	listening_ip,
		const std::string &	client_id,
		const std::filesystem::path &	listening_dir,
		const std::string &	connection_prefix = "snbmodules",
		int	timeout_send = 10,
		int	timeout_receive = 100 )

TransferClient constructor.

Parameters

listening_ip	Listening IP address
port	Listening port
client_id	Client ID
listening_dir	Listening directory (where the client will save incoming files and files to share with bookkeepers)

Definition at line 18 of file transfer_client.cpp.

  : NotificationInterface(connection_prefix, timeout_send, timeout_receive)
  , m_listening_ip(listening_ip)
  , m_client_id(client_id)
{
  TLOG() << "DAB " << __LINE__ << " " << listening_dir;
  // remove all occurences of ./ in the file path
  std::string file_path_str = listening_dir.string();
  std::string x = "./";
 
  size_t pos = 0;
  while (true) {
    TLOG() << "DAB " << __LINE__ << " " << file_path_str;
    pos = file_path_str.find(x, pos);
    if (pos == std::string::npos) {
      break;
    }
 
    file_path_str.replace(pos, x.length(), "");
  }
  TLOG() << "DAB " << __LINE__ << " " << file_path_str;
  if (file_path_str.length() == 0) {
    file_path_str = "blah";
  }
  TLOG() << "DAB " << __LINE__ << " " << file_path_str;
  m_listening_dir = std::filesystem::absolute(file_path_str);
  TLOG() << "DAB " << __LINE__ << " " << file_path_str;
  std::filesystem::create_directories(m_listening_dir);
}

~TransferClient()

dunedaq::snbmodules::TransferClient::~TransferClient

(

)

Definition at line 53 of file transfer_client.cpp.

{}

Member Function Documentation

action_on_receive_notification()

bool dunedaq::snbmodules::TransferClient::action_on_receive_notification ( NotificationData notif )

overrideprivatevirtual

Action to do when the client receive a notification.

Parameters

notif

Notification received

Returns

True if the notification was handled

Implements dunedaq::snbmodules::NotificationInterface.

Definition at line 291 of file transfer_client.cpp.

{
  TLOG() << "debug : Action on request " << notif.m_notification << " for " << notif.m_target_id;
 
  if (notif.m_target_id.find(get_client_id()) == std::string::npos && notif.m_target_id != "all") {
    ers::warning(NotificationWrongDestinationError(ERS_HERE, get_client_id(), notif.m_source_id, notif.m_target_id));
    return false;
  }
 
  // Use enum cast for converting string to enum, still working with older clients and user readable
  auto action = notification_type::string_to_notification(notif.m_notification);
 
  if (action.has_value() == false) {
    ers::warning(InvalidNotificationReceivedError(ERS_HERE, get_client_id(), notif.m_source_id, notif.m_notification));
  }
 
  switch (action.value()) {
 
    case notification_type::e_notification_type::CONNECTION_REQUEST: {
      TLOG() << "debug : receive connection request, sending available files";
      std::set<std::filesystem::path> to_share;
      scan_available_files(to_share, true);
      share_available_files(to_share, notif.m_source_id);
      send_notification(notification_type::e_notification_type::TRANSFER_METADATA,
                        get_client_id(),
                        notif.m_source_id,
                        notif.m_source_id,
                        "end");
      break;
    }
 
    case notification_type::e_notification_type::NEW_TRANSFER: {
      GroupMetadata metadata(notif.m_data, false);
      e_session_type type = Downloader;
 
      // If file is available, create a downloader session
      for (const auto& f : metadata.get_expected_files()) {
        if (m_available_files.find(f) != m_available_files.end()) {
          type = Uploader;
          break;
        }
      }
 
      TLOG() << "debug : creating session " << notif.m_target_id << " type "
             << TransferSession::session_type_to_string(type);
      std::string group_id_tmp = metadata.get_group_id();
      create_session(metadata, type, notif.m_target_id, get_listening_dir().append(group_id_tmp));
      break;
    }
 
    case notification_type::e_notification_type::TRANSFER_METADATA: {
      auto fmeta = std::make_shared<TransferMetadata>(notif.m_data, false);
      auto& sessions_ref = get_sessions();
 
      // print sessions names
      for (auto& s : sessions_ref) {
        TLOG() << "debug : session " << s.get_session_id();
        if (s.get_session_id().find(notif.m_target_id) != std::string::npos) {
          TLOG() << "debug : session found";
          fmeta->set_dest(s.get_ip());
          s.add_file(fmeta);
          return true;
        }
      }
 
      ers::warning(SessionIDNotFoundInClientError(ERS_HERE, get_client_id(), notif.m_target_id));
 
      break;
    }
 
    case notification_type::e_notification_type::START_TRANSFER: {
      TLOG() << "debug : starting transfer " << notif.m_target_id;
      TransferSession* ses = get_session(notif.m_target_id);
      if (ses != nullptr) {
 
        if (notif.m_data == "") {
          ses->start_all();
        } else {
          ses->start_file(ses->get_transfer_options().get_transfer_meta_from_file_path(notif.m_data));
        }
      } else {
        ers::warning(SessionIDNotFoundInClientError(ERS_HERE, get_client_id(), notif.m_target_id));
      }
 
      break;
    }
 
    case notification_type::e_notification_type::PAUSE_TRANSFER: {
      TLOG() << "debug : pausing transfer " << notif.m_target_id;
      TransferSession* ses = get_session(notif.m_target_id);
      if (ses != nullptr) {
 
        if (notif.m_data == "") {
          ses->pause_all();
        } else {
          ses->pause_file(ses->get_transfer_options().get_transfer_meta_from_file_path(notif.m_data));
        }
      } else {
        ers::warning(SessionIDNotFoundInClientError(ERS_HERE, get_client_id(), notif.m_target_id));
      }
 
      break;
    }
 
    case notification_type::e_notification_type::RESUME_TRANSFER: {
      TLOG() << "debug : resuming transfer " << notif.m_target_id;
      TransferSession* ses = get_session(notif.m_target_id);
      if (ses != nullptr) {
 
        if (notif.m_data == "") {
          ses->resume_all();
        } else {
          ses->resume_file(ses->get_transfer_options().get_transfer_meta_from_file_path(notif.m_data));
        }
      } else {
        ers::warning(SessionIDNotFoundInClientError(ERS_HERE, get_client_id(), notif.m_target_id));
      }
 
      break;
    }
 
    case notification_type::e_notification_type::CANCEL_TRANSFER: {
      TLOG() << "debug : cancelling transfer " << notif.m_target_id;
      TransferSession* ses = get_session(notif.m_target_id);
      if (ses != nullptr) {
 
        if (notif.m_data == "") {
          ses->cancel_all();
        } else {
          ses->cancel_file(ses->get_transfer_options().get_transfer_meta_from_file_path(notif.m_data));
        }
      } else {
        ers::warning(SessionIDNotFoundInClientError(ERS_HERE, get_client_id(), notif.m_target_id));
      }
 
      break;
    }
 
    case notification_type::e_notification_type::UPDATE_REQUEST: {
      TLOG() << "debug : updating grp transfer for " << notif.m_target_id;
      TransferSession* ses = get_session(notif.m_target_id);
      if (ses != nullptr) {
 
        if (notif.m_data == "") {
          ses->update_metadatas_to_bookkeeper();
        } else {
          ses->update_metadata_to_bookkeeper(
            ses->get_transfer_options().get_transfer_meta_from_file_path(notif.m_data));
        }
      } else {
        ers::warning(SessionIDNotFoundInClientError(ERS_HERE, get_client_id(), notif.m_target_id));
      }
 
      break;
    }
 
    default:
      ers::warning(NotHandledNotificationError(ERS_HERE, get_client_id(), notif.m_source_id, notif.m_notification));
  }
  return true;
}

cancel_transfer()

void dunedaq::snbmodules::TransferClient::cancel_transfer

(

const std::string &

transfer_id

)

Definition at line 223 of file transfer_client.cpp.

{
 
  TransferSession* session = get_session(transfer_id);
 
  if (session == nullptr) {
    ers::warning(SessionIDNotFoundInClientError(ERS_HERE, get_client_id(), transfer_id));
    return;
  }
 
  session->cancel_all();
}

create_metadata_from_file()

std::shared_ptr< TransferMetadata > dunedaq::snbmodules::TransferClient::create_metadata_from_file ( const std::filesystem::path & src )

private

Create a metadata from a file.

Parameters

src	Path of the file

Returns

Metadata of the file

Definition at line 560 of file transfer_client.cpp.

{
  return std::make_shared<TransferMetadata>(src, std::filesystem::file_size(src), get_ip());
}

create_new_transfer()

void dunedaq::snbmodules::TransferClient::create_new_transfer	(	const std::string &	transfer_id,
		const std::string &	protocol,
		const std::set< std::string > &	dest_clients,
		const std::set< std::filesystem::path > &	files,
		const nlohmann::json &	protocol_options = nlohmann::json() )

Create a new transfer.

Parameters

transfer_id	ID of the transfer
protocol	Protocol to use
dest_clients	Set of destination clients
files	Set of files to transfer
protocol_options	Protocol options

Definition at line 103 of file transfer_client.cpp.

{
 
  std::string session_name = generate_session_id(transfer_id);
 
  // Checking if transfer already exists
  auto ses = get_session(session_name);
  if (ses != nullptr) {
    TLOG() << "debug : transfer " << transfer_id << " already exists !";
    return;
  }
 
  TLOG() << "debug : creating new transfer with protocol " << protocol;
 
  std::optional<protocol_type::e_protocol_type> _protocol = protocol_type::string_to_protocols(protocol);
  if (!_protocol.has_value()) {
    ers::error(InvalidProtocolError(ERS_HERE, get_client_id(), protocol));
    return;
  }
  // Initialize transfer
 
  GroupMetadata group_transfer(transfer_id, session_name, m_listening_ip, _protocol.value(), protocol_options);
 
  for (const auto& file : files) {
    // Check if file exists
    if (!std::filesystem::exists(file)) {
      ers::warning(FileForTransferNotExistError(ERS_HERE, get_client_id(), file));
      continue;
    } else {
      group_transfer.add_expected_file(file);
      group_transfer.add_file(std::move(create_metadata_from_file(file)));
    }
  }
 
  if (group_transfer.get_transfers_meta().empty()) {
    ers::error(FileForTransferNotExistError(ERS_HERE, get_client_id(), "All files"));
    return;
  }
 
  // Sending to bookkeepers to update preparing state
  for (auto& bk : get_bookkeepers_conn()) {
    send_notification(
      notification_type::e_notification_type::GROUP_METADATA, session_name, bk, bk, group_transfer.export_to_string());
  }
 
  // Create local session, can take time depending on protocol
  auto& s = create_session(std::move(group_transfer),
                           e_session_type::Uploader,
                           session_name,
                           get_listening_dir().append(transfer_id),
                           m_listening_ip,
                           dest_clients);
 
  // Notify clients and sending group metadata
  for (const auto& client : dest_clients) {
    std::string new_session_name = generate_session_id(transfer_id, client);
    TLOG() << "debug : notifying client " << client;
    send_notification(notification_type::e_notification_type::NEW_TRANSFER,
                      get_client_id(),
                      new_session_name,
                      client,
                      s.get_transfer_options().export_to_string());
 
    // Send transfer metadata
    for (auto file : s.get_transfer_options().get_transfers_meta()) {
      send_notification(notification_type::e_notification_type::TRANSFER_METADATA,
                        get_client_id(),
                        new_session_name,
                        client,
                        file->export_to_string());
    }
  }
}

create_session()

TransferSession & dunedaq::snbmodules::TransferClient::create_session	(	GroupMetadata	transfer_options,
		e_session_type	type,
		std::string	id,
		const std::filesystem::path &	work_dir,
		IPFormat	ip = IPFormat(),
		const std::set< std::string > &	dest_clients = std::set<std::string>() )

Create a new session, you can precise the IP address of the session forcing to use a specific network interface and different port that the client.

Parameters

transfer_options	Transfer options for the protocol
type	Type of the session (upload or download)
id	ID of the session
listening_ip	Listening IP\:PORT address of the session

Returns

Pointer to the new session

Definition at line 253 of file transfer_client.cpp.

{
  if (ip.is_default()) {
    ip = get_ip();
  }
 
  TransferSession new_session(
    std::move(transfer_options), type, id, ip, work_dir, get_bookkeepers_conn(), get_clients_conn());
 
  m_sessions.emplace_back(std::move(new_session));
  TLOG() << "debug : session created " << TransferSession::session_type_to_string(type);
  m_sessions.back().set_target_clients(dest_clients);
 
  return m_sessions.back();
}

do_work()

bool dunedaq::snbmodules::TransferClient::do_work

(

std::atomic< bool > &

running_flag

)

Start function to use in a thread.

Parameters

running_flag

Flag to stop the client

Returns

False if the client did not stopped correctly

Definition at line 79 of file transfer_client.cpp.

{
 
  while (running_flag.load()) {
    if (get_bookkeepers_conn().size() == 0) {
      TLOG() << "debug : no bookkeeper connected, looking for connections";
      lookups_connections();
    }
    // When starting, client wait for notification from bookkeeper.
    std::optional<NotificationData> msg = listen_for_notification(get_my_conn());
    if (msg.has_value()) {
      action_on_receive_notification(msg.value());
 
      // print status of sessions
      for (const auto& session : m_sessions) {
        TLOG() << session.to_string();
      }
    }
  }
 
  return true;
}

generate_session_id()

std::string dunedaq::snbmodules::TransferClient::generate_session_id ( const std::string & transferid, const std::string & dest = "" )

private

Function to generate session ID.

Definition at line 477 of file transfer_client.cpp.

{
  std::string id = "";
 
  id += dest_id == "" ? get_client_id() : dest_id;
  id += "_ses";
  id += transferid;
 
  return id;
}

get_client_id()

std::string dunedaq::snbmodules::TransferClient::get_client_id ( ) const

inline

Definition at line 105 of file transfer_client.hpp.

{ return m_client_id; }

get_ip()

IPFormat dunedaq::snbmodules::TransferClient::get_ip ( ) const

inline

Definition at line 104 of file transfer_client.hpp.

{ return m_listening_ip; }

get_listening_dir()

std::filesystem::path dunedaq::snbmodules::TransferClient::get_listening_dir ( ) const

inline

Definition at line 106 of file transfer_client.hpp.

{ return m_listening_dir; }

get_my_conn()

std::string dunedaq::snbmodules::TransferClient::get_my_conn

(

)

Definition at line 566 of file transfer_client.cpp.

{
  TLOG() << "BBB " << __LINE__ << " my_conn=\"" << m_my_conn << "\"";
  if (m_my_conn.empty()) {
    for (const std::string& c : get_clients_conn()) {
      std::string str1 = c;
      std::string str2 = get_client_id();
 
      if (str1.starts_with("snbmodules_")) {
        str1 = str1.substr(11);
      }
      if (str2.starts_with("snbmodules_")) {
        str2 = str2.substr(11);
      }
 
      if (str1.starts_with("snb-sample-config-")) {
        str1 = str1.substr(18);
      }
      if (str2.starts_with("snb-sample-config-")) {
        str2 = str2.substr(18);
      }
 
      TLOG() << "BBB " << __LINE__ << " c=\"" << c << "\" client_id=" << get_client_id() << " " << str1 << " " << str2;
      // if (c.find(get_client_id()) != std::string::npos)
      if (str1 == str2) {
        m_my_conn = c;
        break;
      }
    }
 
    if (m_my_conn.empty()) {
      ers::warning(ConnectionNotFoundError(ERS_HERE, get_client_id()));
    }
  }
  return m_my_conn;
}

get_session()

TransferSession * dunedaq::snbmodules::TransferClient::get_session

(

std::string

transfer_id

)

Definition at line 237 of file transfer_client.cpp.

{
  if (transfer_id.find("ses") == std::string::npos) {
    transfer_id = generate_session_id(transfer_id);
  }
 
  for (auto& s : m_sessions) {
    if (s.get_session_id() == transfer_id) {
      return &s;
    }
  }
  return nullptr;
}

get_sessions() [1/2]

std::vector< TransferSession > & dunedaq::snbmodules::TransferClient::get_sessions ( )

inline

Definition at line 108 of file transfer_client.hpp.

{ return m_sessions; }

get_sessions() [2/2]

const std::vector< TransferSession > & dunedaq::snbmodules::TransferClient::get_sessions ( ) const

inline

Definition at line 109 of file transfer_client.hpp.

{ return m_sessions; }

pause_transfer()

void dunedaq::snbmodules::TransferClient::pause_transfer

(

const std::string &

transfer_id

)

Definition at line 196 of file transfer_client.cpp.

{
 
  TransferSession* session = get_session(transfer_id);
 
  if (session == nullptr) {
    ers::warning(SessionIDNotFoundInClientError(ERS_HERE, get_client_id(), transfer_id));
    return;
  }
 
  session->pause_all();
}

remove_session()

void dunedaq::snbmodules::TransferClient::remove_session ( const std::string & session_id )

private

Add a session to the client.

Parameters

session

Session to add

Remove a session from the client

Parameters

session_id

ID of the session to remove

Definition at line 461 of file transfer_client.cpp.

{
  auto s = get_session(session_id);
  if (s == nullptr) {
    ers::warning(SessionIDNotFoundInClientError(ERS_HERE, get_client_id(), session_id));
    return;
  }
  for (size_t i = 0; i < m_sessions.size(); i++) {
    if (s->get_session_id() == session_id) {
      m_sessions.erase(m_sessions.begin() + i);
      return;
    }
  }
}

resume_transfer()

void dunedaq::snbmodules::TransferClient::resume_transfer

(

const std::string &

transfer_id

)

Definition at line 209 of file transfer_client.cpp.

{
 
  TransferSession* session = get_session(transfer_id);
 
  if (session == nullptr) {
    ers::warning(SessionIDNotFoundInClientError(ERS_HERE, get_client_id(), transfer_id));
    return;
  }
 
  session->resume_all();
}

scan_available_files()

void dunedaq::snbmodules::TransferClient::scan_available_files	(	std::set< std::filesystem::path > &	previous_scan,
		bool	nested = false,
		std::filesystem::path	folder = std::filesystem::path() )

Scan available files in the listening directory.

Parameters

previous_scan	Set of files already scanned
folder	Folder to scan
nested	True if the function is called recursively

Definition at line 489 of file transfer_client.cpp.

{
  if (folder.empty()) {
    folder = get_listening_dir();
  }
 
  TLOG() << "debug : scanning files in " << folder;
 
  // First scan group metadata files
  for (const auto& entry : std::filesystem::directory_iterator(folder)) {
    if (entry.is_regular_file() && entry.path().extension() == GroupMetadata::m_file_extension) {
      if (previous_scan.insert(entry.path()).second) {
        TLOG() << "debug : found new file " << entry.path();
 
        GroupMetadata metadata = GroupMetadata(entry.path());
        bool already_active = false;
 
        for (const auto& s : get_sessions()) {
          if (s.get_transfer_options() == metadata) {
            already_active = true;
            break;
          }
        }
 
        if (!already_active) {
          std::string group_id_tmp = metadata.get_group_id();
          create_session(metadata,
                         Uploader,
                         generate_session_id(group_id_tmp),
                         get_listening_dir().append("ses" + std::to_string(m_sessions.size())));
        }
      }
    }
  }
 
  for (const auto& entry : std::filesystem::directory_iterator(folder)) {
    if (entry.is_regular_file() && entry.path().extension() != GroupMetadata::m_file_extension) {
      if (previous_scan.insert(entry.path()).second) {
        TLOG() << "debug : found new file " << entry.path();
        if (entry.path().extension() == TransferMetadata::m_file_extension) {
          std::shared_ptr<TransferMetadata> metadata = std::make_shared<TransferMetadata>(entry.path());
 
          bool wanted = false;
          for (auto& s : get_sessions()) {
            auto expect_ref = s.get_transfer_options().get_expected_files();
            if (expect_ref.find(metadata->get_file_name()) != expect_ref.end()) {
              s.add_file(metadata);
              wanted = true;
              break;
            }
          }
          if (!wanted) {
            // If not wanted yet (or never), delete the file from the scan list to find it again later
            TLOG() << "debug : " << entry.path() << " not wanted yet, deleting from scan list";
            previous_scan.erase(entry.path());
            metadata.reset();
          }
        }
      }
    } else if (entry.is_directory()) {
      TLOG() << "debug : found directory " << entry.path();
      if (nested) {
        scan_available_files(previous_scan, nested, entry.path()); // NOLINT
      }
    }
  }
}

set_client_id()

void dunedaq::snbmodules::TransferClient::set_client_id ( std::string client_id )

inline

Definition at line 115 of file transfer_client.hpp.

{ m_client_id = std::move(client_id); }

set_ip()

void dunedaq::snbmodules::TransferClient::set_ip ( const std::string & ip )

inline

Definition at line 113 of file transfer_client.hpp.

{ m_listening_ip.set_ip(ip); }

set_listening_dir()

void dunedaq::snbmodules::TransferClient::set_listening_dir ( const std::filesystem::path & listening_dir )

inline

Definition at line 116 of file transfer_client.hpp.

  {
    // remove all occurences of ./ in the file path
    std::string file_path_str = listening_dir.string();
    std::string x = "./";
 
    size_t pos = 0;
    while (true) {
      pos = file_path_str.find(x, pos);
      if (pos == std::string::npos) {
        break;
      }
 
      file_path_str.replace(pos, x.length(), "");
    }
 
    m_listening_dir = std::filesystem::absolute(file_path_str);
  }

set_port()

void dunedaq::snbmodules::TransferClient::set_port ( int port )

inline

Definition at line 114 of file transfer_client.hpp.

{ m_listening_ip.set_port(port); }

share_available_files()

void dunedaq::snbmodules::TransferClient::share_available_files ( const std::set< std::filesystem::path > & to_share, const std::string & dest )

private

Share available files (in m_listening_dir)

Parameters

to_share	Set of files to share
dest	Destination of the files

Definition at line 275 of file transfer_client.cpp.

{
 
  for (const std::filesystem::path& f : to_share) {
    TLOG() << "debug : Sharing " << f.filename();
    std::shared_ptr<TransferMetadata> fmeta = create_metadata_from_file(f);
    m_available_files.insert({ f.string(), fmeta });
    send_notification(notification_type::e_notification_type::TRANSFER_METADATA,
                      get_client_id(),
                      dest,
                      dest,
                      fmeta->export_to_string());
  }
}

start()

bool dunedaq::snbmodules::TransferClient::start

(

int

timeout

)

Start the client in curent thread.

Parameters

timeout

Timeout in seconds, after this time the client will stop

Returns

False if the client did not stopped correctly

Definition at line 56 of file transfer_client.cpp.

{
 
  while (true) {
    // When starting, client wait for notification from bookkeeper.
    std::optional<NotificationData> msg = listen_for_notification(get_my_conn(), "", timeout);
    if (msg.has_value()) {
      action_on_receive_notification(msg.value());
    } else {
      TLOG() << "debug : no notification received, timeout";
      return false;
    }
 
    // print status of sessions
    for (const auto& session : m_sessions) {
      TLOG() << session.to_string();
    }
  }
 
  return true;
}

start_transfer()

void dunedaq::snbmodules::TransferClient::start_transfer

(

const std::string &

transfer_id

)

Start, pause, resume or cancel a transfer.

Parameters

transfer_id

ID of the transfer to start, pause, resume or cancel

Definition at line 182 of file transfer_client.cpp.

{
 
  TransferSession* session = get_session(transfer_id);
 
  if (session == nullptr) {
    ers::warning(SessionIDNotFoundInClientError(ERS_HERE, get_client_id(), transfer_id));
    return;
  }
 
  session->start_all();
}

Member Data Documentation

m_available_files

std::map<std::string, std::shared_ptr<TransferMetadata> > dunedaq::snbmodules::TransferClient::m_available_files

private

Map of available files (key = file path, value = file metadata)

Definition at line 149 of file transfer_client.hpp.

m_client_id

std::string dunedaq::snbmodules::TransferClient::m_client_id

private

TransferClient ID, unique identifier of the client.

Definition at line 140 of file transfer_client.hpp.

m_listening_dir

std::filesystem::path dunedaq::snbmodules::TransferClient::m_listening_dir

private

Listening directory, directory where the client will listen for incoming files and files to share.

Definition at line 143 of file transfer_client.hpp.

m_listening_ip

IPFormat dunedaq::snbmodules::TransferClient::m_listening_ip

private

IP address of the client.

Definition at line 137 of file transfer_client.hpp.

m_my_conn

std::string dunedaq::snbmodules::TransferClient::m_my_conn = ""

private

Connection uuid of the client, retrieved using the notification interface and calling get_my_conn()

Definition at line 152 of file transfer_client.hpp.

m_sessions

std::vector<TransferSession> dunedaq::snbmodules::TransferClient::m_sessions

private

Map of active sessions (key = session ID, value = session)

Definition at line 146 of file transfer_client.hpp.

---

The documentation for this class was generated from the following files:

• /github/workspace/dunedaq/sourcecode/snbmodules/include/snbmodules/transfer_client.hpp
• /github/workspace/dunedaq/sourcecode/snbmodules/src/client/transfer_client.cpp