I have a dbus (linux) program in C that send a signal with a destination. The code is the following:

#include <stdio.h>
#include <stdlib.h>
#include <dbus/dbus.h>
#include <unistd.h>

/**
 * Connect to the DBUS bus and send a broadcast signal
 */
void sendsignal(DBusConnection* conn, char* sigvalue)
{
   DBusMessage* msg;
   DBusMessageIter args;
   DBusError err;
   int ret;
   dbus_uint32_t serial = 0;

   printf("Sending signal with value %s\n", sigvalue);

   // create a signal & check for errors 
   msg = dbus_message_new_signal("/test/signal/Object", // object name of the signal
                                 "test.signal.Type", // interface name of the signal
                                 "Test"); // name of the signal
   if (NULL == msg) 
   { 
      fprintf(stderr, "Message Null\n"); 
      exit(1); 
   }

   // set the destination of the signal
   dbus_message_set_destination(msg, "it.smartsecurity.dbus");


   // append arguments onto signal
   dbus_message_iter_init_append(msg, &args);
   if (!dbus_message_iter_append_basic(&args, DBUS_TYPE_STRING, &sigvalue)) {
      fprintf(stderr, "Out Of Memory!\n"); 
      exit(1);
   }

   // send the message and flush the connection
   if (!dbus_connection_send(conn, msg, &serial)) {
      fprintf(stderr, "Out Of Memory!\n"); 
      exit(1);
   }

   dbus_connection_flush(conn);
   
   printf("Signal Sent\n");
   
   // free the message and close the connection
   dbus_message_unref(msg);
   //dbus_connection_close(conn);
}


int main(int argc, char **argv) {

   DBusMessage* msg;
   DBusMessageIter args;
   DBusConnection* conn;
   DBusError err; 
   int ret;

   // initialise the error value
   dbus_error_init(&err);

   // connect to the DBUS system bus, and check for errors
   conn = dbus_bus_get(DBUS_BUS_SYSTEM, &err);
   if (dbus_error_is_set(&err)) { 
      fprintf(stderr, "Connection Error (%s)\n", err.message); 
      dbus_error_free(&err); 
   }
   if (NULL == conn) { 
      exit(1); 
   }

   // register our name on the bus, and check for errors
   ret = dbus_bus_request_name(conn, "it.smartsecurity.dbus", DBUS_NAME_FLAG_REPLACE_EXISTING , &err);
   if (dbus_error_is_set(&err)) { 
      fprintf(stderr, "Name Error (%s)\n", err.message); 
      dbus_error_free(&err); 
   }
   if (DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER != ret) { 
      exit(1);
   }

   do {
      
      sendsignal(conn, "CIAO");
      sleep(1);
      
   } while (1);

   // dbus_connection_close(conn);
   return 0;
}

Then with zbus crate I wrote the followng code:

use futures_util::stream::StreamExt;
use zbus::{zvariant::OwnedObjectPath, proxy, Connection};
use zbus::zvariant::Value;


#[proxy(
    default_service = "it.smartsecurity.dbus",
    default_path = "/test/signal/Object",
    interface = "test.signal.Type",
)]
trait Systemd1Manager {
    // Defines signature for D-Bus signal named `Test`
    #[zbus(signal)]
    fn test(&self, unit: String) -> zbus::Result<()>; // si deve chiamare come il segnale
}

async fn watch_systemd_jobs() -> zbus::Result<()> {
    let connection = Connection::system().await?;
    // `Systemd1ManagerProxy` is generated from `Systemd1Manager` trait
  
    let systemd_proxy = Systemd1ManagerProxy::builder(&connection)
        .destination("it.smartsecurity.dbus")?
        .path("/test/signal/Object")?
        .interface("test.signal.Type")?
        .build().await?;

    // Method `receive_job_new` is generated from `job_new` signal
    let mut new_jobs_stream = systemd_proxy.receive_test().await?;

    while let Some(msg) = new_jobs_stream.next().await {
        //dbg!(&msg);
        // struct `JobNewArgs` is generated from `job_new` signal function arguments
        let args = msg.args();
        dbg!(&args);

        println!("=====================");
        // stampa il nome del servizio e il suo valore
        let x = msg.message().header();
        
        let y = x.member();
        if y.is_some() {
            println!("Header: {}", y.unwrap());
        }

        dbg!(&y);

        let unit = args.unwrap().unit;
        println!("Param: {}", unit);

    }

    panic!("Stream ended unexpectedly");
}

#[tokio::main]
async fn main() {
    watch_systemd_jobs().await.unwrap();
}

My problem is that this Rust code doesn't catch the signal and I don't understand what I'm doing wrong.

I have a dbus (linux) program in C that send a signal with a destination. The code is the following:

#include <stdio.h>
#include <stdlib.h>
#include <dbus/dbus.h>
#include <unistd.h>

/**
 * Connect to the DBUS bus and send a broadcast signal
 */
void sendsignal(DBusConnection* conn, char* sigvalue)
{
   DBusMessage* msg;
   DBusMessageIter args;
   DBusError err;
   int ret;
   dbus_uint32_t serial = 0;

   printf("Sending signal with value %s\n", sigvalue);

   // create a signal & check for errors 
   msg = dbus_message_new_signal("/test/signal/Object", // object name of the signal
                                 "test.signal.Type", // interface name of the signal
                                 "Test"); // name of the signal
   if (NULL == msg) 
   { 
      fprintf(stderr, "Message Null\n"); 
      exit(1); 
   }

   // set the destination of the signal
   dbus_message_set_destination(msg, "it.smartsecurity.dbus");


   // append arguments onto signal
   dbus_message_iter_init_append(msg, &args);
   if (!dbus_message_iter_append_basic(&args, DBUS_TYPE_STRING, &sigvalue)) {
      fprintf(stderr, "Out Of Memory!\n"); 
      exit(1);
   }

   // send the message and flush the connection
   if (!dbus_connection_send(conn, msg, &serial)) {
      fprintf(stderr, "Out Of Memory!\n"); 
      exit(1);
   }

   dbus_connection_flush(conn);
   
   printf("Signal Sent\n");
   
   // free the message and close the connection
   dbus_message_unref(msg);
   //dbus_connection_close(conn);
}


int main(int argc, char **argv) {

   DBusMessage* msg;
   DBusMessageIter args;
   DBusConnection* conn;
   DBusError err; 
   int ret;

   // initialise the error value
   dbus_error_init(&err);

   // connect to the DBUS system bus, and check for errors
   conn = dbus_bus_get(DBUS_BUS_SYSTEM, &err);
   if (dbus_error_is_set(&err)) { 
      fprintf(stderr, "Connection Error (%s)\n", err.message); 
      dbus_error_free(&err); 
   }
   if (NULL == conn) { 
      exit(1); 
   }

   // register our name on the bus, and check for errors
   ret = dbus_bus_request_name(conn, "it.smartsecurity.dbus", DBUS_NAME_FLAG_REPLACE_EXISTING , &err);
   if (dbus_error_is_set(&err)) { 
      fprintf(stderr, "Name Error (%s)\n", err.message); 
      dbus_error_free(&err); 
   }
   if (DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER != ret) { 
      exit(1);
   }

   do {
      
      sendsignal(conn, "CIAO");
      sleep(1);
      
   } while (1);

   // dbus_connection_close(conn);
   return 0;
}

Then with zbus crate I wrote the followng code:

use futures_util::stream::StreamExt;
use zbus::{zvariant::OwnedObjectPath, proxy, Connection};
use zbus::zvariant::Value;


#[proxy(
    default_service = "it.smartsecurity.dbus",
    default_path = "/test/signal/Object",
    interface = "test.signal.Type",
)]
trait Systemd1Manager {
    // Defines signature for D-Bus signal named `Test`
    #[zbus(signal)]
    fn test(&self, unit: String) -> zbus::Result<()>; // si deve chiamare come il segnale
}

async fn watch_systemd_jobs() -> zbus::Result<()> {
    let connection = Connection::system().await?;
    // `Systemd1ManagerProxy` is generated from `Systemd1Manager` trait
  
    let systemd_proxy = Systemd1ManagerProxy::builder(&connection)
        .destination("it.smartsecurity.dbus")?
        .path("/test/signal/Object")?
        .interface("test.signal.Type")?
        .build().await?;

    // Method `receive_job_new` is generated from `job_new` signal
    let mut new_jobs_stream = systemd_proxy.receive_test().await?;

    while let Some(msg) = new_jobs_stream.next().await {
        //dbg!(&msg);
        // struct `JobNewArgs` is generated from `job_new` signal function arguments
        let args = msg.args();
        dbg!(&args);

        println!("=====================");
        // stampa il nome del servizio e il suo valore
        let x = msg.message().header();
        
        let y = x.member();
        if y.is_some() {
            println!("Header: {}", y.unwrap());
        }

        dbg!(&y);

        let unit = args.unwrap().unit;
        println!("Param: {}", unit);

    }

    panic!("Stream ended unexpectedly");
}

#[tokio::main]
async fn main() {
    watch_systemd_jobs().await.unwrap();
}

My problem is that this Rust code doesn't catch the signal and I don't understand what I'm doing wrong.

• c
• rust
• dbus

1 Answer 1

When you send bus messages to a specific destination, then they are no longer broadcast (contrary to your code comments) – they are only delivered to that destination. Only the process which has claimed the bus name it.smartsecurity.dbus will be able to receive such signals.

In your code, the bus name is claimed by the sender rather than the receiver, and therefore the sender is actually sending the signals to itself.