[Solved] API harmonisation

[Mallets]

The following covers some aspects about rationalisation of APIs, their semantics, and implied behaviour for Rust, C, and Python.

Philosophy

Properly naming the functions has been a great ground for debate. In the following I propose a common philosophy for function naming and the implied behaviour.

Concept

Let’s start by categorising the zenoh session functions in two sets:

• Declaration
• Action

Declarations

Functions that perform a certain declaration is expected to have a long-lived impact on the zenoh system to a certain degree. E.g. some permanent state is maintained in the network, information is distributed, some mechanism is enabled in the system.

That means, when performing a declaration, the zenoh system (either next hop or far away) will potentially react and adapt its behaviour according to the declaration. As such, the amount of declarations in a given system needs to be maintained into reasonable limits. Moreover, declarations represent something that can be reverted. I.e., if something can be declared then it can also be undeclared. Finally, nouns should be used to name declaration functions.

Example of declarations are:

• subscriber
• publisher
• queryable
• querier
• keyexpr

Actions

Functions that perform a certain action is expected to have a none or short-living impact on the zenoh system to a certain degree. That means, when performing an action the zenoh system is expected to simply deliver or retrieve data. As such, a given zenoh system should be able to support a large number of concurrent actions. In other words, we could refer to actions as the set of functions that belong to the critical path. Moreover, actions represent some action that can be reverted per-se. I.e., a put on the network can not be reverted but a second action like delete can be provided to undo the effects of a put at a certain degree. Finally, verbs should be used to name action functions.

Example of actions are:

• put
• delete
• get
• reply
• pull

NOTE: in the current API version (0.5.0-beta.9) the declare_ is intended to mark optional zenoh functions. In the version proposed below, the term/function declare is instead used for marking a declaration.

Future extensibility

It is desirable to keep in the future the possibility to extend the current API with additional parameters and functionalities. However, this shouldn’t have an impact on the function signature and it should be rather handled in a future-proof manner.

Depending on the programming language, future extensibility can be provided by three paradigms:

• builder pattern
  • the declaration/action returns a builder that needs to be resolved to perform an action. E.g., this approach is suitable for Rust.
• kwargs with default values
  • this approach practically embeds a builder in the form of extensible signature where more and more parameters can be added in the future. E.g. this approach is suitable for Python and Kotlin.
• option struct
  • the declaration/action signature always takes a struct that contains all the options. A default option constructor is provided to the user. E.g. this approach is suitable for C.

Sync and Async

It has been debated wether sync and async API should be provided separately or by the same type.
The main goal is to avoid the uncareful user to mix sync and async calls if not expressly desired.
More discussion can be found here: ZFuture.
The proposal below it hence adopts the declaration and action separation together with the sync and async separated APIs.

NOTE: additional information will be posted shortly

[Mallets]

Declaration

A zenoh session allows to perform a set of declarations to interact with the overall zenoh system. Declarations are by nature revertible, i.e. everything that is being declared can be undeclared at a later stage. In the same way as declarations are done on the session, undeclarations are done on the same session.

Rust

Below you can find a common approach to declarations in Rust. Below you can find an example based on the current 0.5.0-beta.9 implementation and types.

trait ZDeclaration {
    fn undeclare_inner(self, session: &Session) -> ZResult<()>;
}

impl<'a> ZDeclaration for KeyExpr<'a> {
    fn undeclare_inner(self, session: &Session) -> ZResult<()> {
        session.undeclare_expr(self.as_id()).wait()
    }
}

impl<'a> ZDeclaration for Subscriber<'a> {
    fn undeclare_inner(mut self, session: &Session) -> ZResult<()> {
        self.alive = false;
        session.unsubscribe(self.state.id).wait()
    }
}

impl<'a> ZDeclaration for Queryable<'a> {
    fn undeclare_inner(mut self, session: &Session) -> ZResult<()> {
        self.alive = false;
        session.close_queryable(self.state.id).wait()
    }
}

impl Session {
    pub fn undeclare<'a, D>(&'a self, d: D) -> ZResult<()>
    where
        D: ZDeclaration,
    {
        d.undeclare_inner(&self)
    }
}

By doing so, any declaration type needs to implement the ZDeclaration trait that allows any declaration to be undeclared via the session regardless its type. It’s one single API that works for all declarations.

session.undeclare(subscriber).unwrap();
session.undeclare(publisher).unwrap();
session.undeclare(queryable).unwrap();
session.undeclare(querier).unwrap();
session.undeclare(keyexpr).unwrap();

Additionally, for those types that carry internally a reference to the session, the undeclare() function could be also provided by the type itself.

subscriber.undeclare().unwrap();
publisher.undeclare().unwrap();
queryable.undeclare().unwrap();
querier.undeclare().unwrap();

Subscriber

A subscriber is a declaration. Therefore, the function should refer to the subscriber noun.

Rust

In Rust, the builder pattern is resolved by the function res(). This applies to any declaration or action.

pub fn declare_subscriber<'a, 'b, IntoKeyExpr>(
    &'a self,
    key_expr: IntoKeyExpr,
) -> SubscriberBuilder<'a, 'b>
where
    IntoKeyExpr: Into<KeyExpr<'b>>;

// --- SYNC
use zenoh::sync::*;

// Callback
let s = session.declare_subscriber("/a").callback(|s| {}).res().unwrap();
// s does not implement .next()
s.undeclare().unwrap(); // shorthand for session.undeclare(s).unwrap();
// Stream
let s = session.declare_subscriber("/a").channel(flume::bounded(64)).res().unwrap();
// s implements .next()
s.undeclare().unwrap(); // shorthand for session.undeclare(s).unwrap();

// Default shorthand for session.subscribe("/a").channel(flume::bounded(64))
let s = session.declare_subscriber("/a").res().unwrap();
s.undeclare().unwrap(); // shorthand for session.undeclare(s).unwrap();

// --- ASYNC
use zenoh::async::*;

// Callback
let s = session.declare_subscriber("/a").callback(|s| async {}).res().await.unwrap();
s.undeclare().await.unwrap(); // shorthand for session.undeclare(s).await.unwrap();
// Stream
let s = session.declare_subscriber("/a").channel(flume::bounded(64)).res().await.unwrap();
s.undeclare().await.unwrap(); // shorthand for session.undeclare(s).await.unwrap();

// Default shorthand for session.subscribe("/a").channel(flume::bounded(64))
let s = session.declare_subscriber("/a").res().await.unwrap();
s.undeclare().await.unwrap(); // shorthand for session.undeclare(s).await.unwrap();

C

In C, an iterator abstraction will be built departing from the callback-based API.

z_owned_subscriber_t sub = z_declare_subscriber(z_loan(s), 
		z_expr("/a"), data_handler, NULL, z_sopt_default());
z_undeclare_subscriber(z_move(sub));

Python

# --- SYNC
import zenoh.sync as zenoh

sub = session.declare_subscriber(key)
sub.undeclare() # shorthand for session.undeclare(sub)

# --- ASYNC
import zenoh.async as zenoh

sub = await session.declare_subscriber(key)
await sub.undeclare() # shorthand for await session.undeclare(sub)

Publisher

A publisher is a declaration. Therefore, the function should refer to the publisher noun.

Rust

pub fn declare_publisher<'a, 'b, IntoKeyExpr>(
    &'a self,
    key_expr: IntoKeyExpr,
) -> PublisherBuilder<'a, 'b>
where
    IntoKeyExpr: Into<KeyExpr<'b>>;

// --- SYNC
use zenoh::sync::*;

// Default
let p = session.declare_publisher("/a").res().unwrap();
loop { 
	p.put(data).unwrap(); 
}
// s does not implement .next()
p.undeclare().unwrap(); // shorthand for session.undeclare(p).unwrap();

// --- ASYNC
use zenoh::async::*;

// Default
let p = session.declare_publisher("/a").res().await.unwrap();
loop { 
	p.put(data).await.unwrap(); 
}
// s does not implement .next()
p.undeclare().await.unwrap(); // shorthand for session.undeclare(p).unwrap();

C

z_owned_publisher_t pub = z_declare_publisher(z_loan(s), 
		z_expr("/a"), z_popt_default());
z_put_pub(z_loan(pub), data);
z_undeclare_publisher(z_move(sub));

Python

# --- SYNC
import zenoh.sync as zenoh

pub = session.declare_publisher(key)
pub.put(data)
pub.undeclare() # shorthand for session.undeclare(sub)

# --- ASYNC
import zenoh.async as zenoh

pub = await session.declare_publisher(key)
await pub.put(data)
await pub.undeclare() # shorthand for await session.undeclare(sub)

Queryable

A queryable is a declaration. Therefore, the function should refer to the queryable noun.

Rust

pub fn declare_queryable<'a, 'b, IntoKeyExpr>(
    &'a self,
    key_expr: IntoKeyExpr,
) -> QueryableBuilder<'a, 'b>
where
    IntoKeyExpr: Into<KeyExpr<'b>>;

// --- SYNC
use zenoh::sync::*;

// Callback
let q = session.declare_queryable("/a").callback(|s| {}).res().unwrap();
// q does not implement .next()
q.undeclare().unwrap(); // shorthand for session.undeclare(q).unwrap();
// Stream
let q = session.declare_queryable("/a").channel(flume::bounded(64)).res().unwrap();
// q implements .next()
q.undeclare().unwrap(); // shorthand for session.undeclare(q).unwrap();

// Default shorthand for session.subscribe("/a").channel(flume::bounded(64))
let q = session.declare_queryable("/a").res().unwrap();
q.undeclare().unwrap(); // shorthand for session.undeclare(q).unwrap();

// --- ASYNC
use zenoh::async::*;

// Callback
let q = session.declare_queryable("/a").callback(|s| async {}).res().await.unwrap();
q.undeclare().await.unwrap(); // shorthand for session.undeclare(q).await.unwrap();
// Stream
let q = session.declare_queryable("/a").channel(flume::bounded(64)).res().await.unwrap();
q.undeclare().await.unwrap(); // shorthand for session.undeclare(q).await.unwrap();

// Default shorthand for session.subscribe("/a").channel(flume::bounded(64))
let q = session.declare_queryable("/a").res().await.unwrap();
q.undeclare().await.unwrap(); // shorthand for session.undeclare(q).await.unwrap();

C

In C, an iterator abstraction will be built departing from the callback-based API.

z_owned_queryable_t qbl = z_declare_queryable(z_loan(s), 
		z_keyexpr("/a"), query_handler, NULL, z_qopt_default());
z_undeclare_queryable(z_move(qbl));

Python

# --- SYNC
import zenoh.sync as zenoh

qbl = session.declare_queryable(key)
qbl.undeclare() # shorthand for session.undeclare(qbl)

# --- ASYNC
import zenoh.async as zenoh

qbl = await session.declare_queryable(key)
await qbl.undeclare() # shorthand for await session.undeclare(qbl)

Querier

A querier is a declaration. Therefore, the function should refer to the querier noun.

Rust

pub fn declare_querier<'a, 'b, IntoKeyExpr>(
    &'a self,
    key_expr: IntoKeyExpr,
) -> QuerierBuilder<'a, 'b>
where
    IntoKeyExpr: Into<KeyExpr<'b>>;

// --- SYNC
use zenoh::sync::*;

// Default
let q = session.declare_querier("/a").res().unwrap();
loop { 
	while let Some(r) = q.get() {
		match r {
			Ok(r) => {},
			Err(e) => {},
		}
  } 
}
// s does not implement .next()
q.undeclare().unwrap(); // shorthand for session.undeclare(q).unwrap();

// --- ASYNC
use zenoh::async::*;

// Default
let q = session.declare_querier("/a").res().await.unwrap();
loop { 
	while let Some(r) = q.get().await {
		match r {
			Ok(r) => {},
			Err(e) => {},
		}
  } 
}
// s does not implement .next()
q.undeclare().await.unwrap(); // shorthand for session.undeclare(q).unwrap();

C

z_owned_querier_t pub = z_declare_querier(z_loan(s), 
		z_keyexpr("/a"), z_qopt_default());
z_get_query(z_loan(pub), data);
z_undeclare_publisher(z_move(sub));

Python

# --- SYNC
import zenoh.sync as zenoh

pub = session.declare_publisher(key)
pub.put(data)
pub.undeclare() # shorthand for session.undeclare(sub)

# --- ASYNC
import zenoh.async as zenoh

pub = await session.declare_publisher(key)
await pub.put(data)
await pub.undeclare() # shorthand for await session.undeclare(sub)

KeyExpr

An expression is a declaration. Therefore, the function should refer to the expr noun.

Rust

Despite the ongoing discussion mentioned above, for what concerns the Rust API, it should be aligned with the other API and potentially make the KeyExpr opaque to the user.

pub fn keyexpr<'a, IntoKeyExpr>(&'a self, key_expr: IntoKeyExpr) -> KeyExprBuilder<'a>
where
    IntoKeyExpr: Into<KeyExpr<'a>>;

// --- SYNC
use zenoh::sync::*;

let k = session.declare_keyexpr("/a").res().unwrap();
session.undeclare(k).unwrap();

// --- ASYNC
use zenoh::async::*;

let k = session.declare_keyexpr("/a").res().await.unwrap();
session.undeclare(k).unwrap();

C

z_keyexpr_t k = z_declare_keyexpr(s, z_keyexpr("/a"), z_eopt_default());
int r = z_undeclare_keyexpr(s, k);

Python

# --- SYNC
import zenoh.sync as zenoh

k = session.declare_keyexpr("/a")
session.undeclare(k)

# --- ASYNC
import zenoh.async as zenoh

k = await session.declare_keyexpr("/a")
await session.undeclare(k)

NOTE: additional information will be posted shortly

[gabrik]

@Mallets quick practical question on the rust implementation for KeyExpr.
Let's imagine that I have a shared Zenoh session (inside an Arc) between different objects, then each one of these objects declares a KeyExpr and stores it in Self, it will be possible to do so? Or the declared KeyExpr will have a reference to the session and therefore a lifetime bound to it?

Let me write some code:

pub struct Something {
    session: Arc<Session>,
    key_expr: zenoh::KeyExpr,
}

impl Something {
   pub fn new(s: Arc<Session>, info: WhatheverType) -> Result<Self, Error> {
      let key = my_super_duper_key_expr_generator(info);
      let key_expr = session.declare_keyexpr(&key).res().wait()?;

     Ok(Self {session: s, key_expr})
   }
   
   pub async fn iter(&self, data: Data) -> Result<()> {
       let serialized = Data.serialize();
       self.session.put(&self.key_expr, data).await?
   }
}

It will be possible to write this code?

[p-avital]

Good question... With it's current implementation, the new KeyExpr is independent from the session that declared it lifetime-wise. That does mean future implementations may need to maintain this should they work by borrowing part of the session.

I think tying the lifetime of the KeyExpr to the session would be too complicated to handle in useful scenarios, so I'd rather we keep the returned KeyExpr 's lifetime decoupled from that of the session reference.

[gabrik]

Thank you @p-avital for the clarification.

I indeed think that having the KeyExpr lifetime being independent of session one will facilitate users.

[Mallets]

Actions

Below you can find a common approach to actions in Rust. Below you can find an example based on the current implementation and types.

Put

A put is an action. Therefore, the function should refer to the put verb.

Rust

pub fn put<'a, IntoKeyExpr, IntoValue>(
    &'a self,
    key_expr: IntoKeyExpr,
    value: IntoValue,
) -> PutBuilder<'a>
where
    IntoKeyExpr: Into<KeyExpr<'a>>,
    IntoValue: Into<Value>;

// --- SYNC
use zenoh::sync::*;

// Default parameters
session.put("/a", data).res().unwrap();

// Optional parameters
session
	.put("/a", data)
	.congestion_control(CongestionControl::Block)
	.res()
	.unwrap();

// --- ASYNC
use zenoh::async::*;

// Default parameters
session.put("/a", data).res().await.unwrap();

// Optional parameters
session
	.put("/a", data)
	.congestion_control(CongestionControl::Block)
  	.res()
 	.await
	.unwrap();

C

int r = z_put(s, keyexpr, (const uint8_t *)data, len, z_popt_default());

Python

# --- SYNC
import zenoh.sync as zenoh

session.put(key, value)

# --- ASYNC
import zenoh.async as zenoh

await session.put(key, value)

Get

A get is an action. Therefore, the function should refer to the get verb.

Rust

pub fn get<'a, 'b, IntoSelector>(&'a self, selector: IntoSelector) -> GetBuilder<'a, 'b>
where
    IntoSelector: Into<Selector<'b>>;

// --- SYNC
use zenoh::sync::*;

// Default: stream;
let r = session.get("/a").res().unwrap();

// Stream
let r = session
	.get("/a")
	.channel(flume::bounded(64))
	.res()
	.unwrap();

// Callback
session
	.get("/a")
	.callback(|s| {})
	.res()
	.unwrap();

// --- ASYNC
use zenoh::async::*;

// Default: stream;
let r = session.get("/a").res().await.unwrap();

// Stream
let r = session
	.get("/a")
	.channel(flume::bounded(64))
	.res()
	.await
	.unwrap();

// Callback
session
	.get("/a")
	.callback(|s| {})
	.res()
	.await
	.unwrap();

C

In C, an iterator abstraction will be built departing from the callback-based API.

int r = z_get(s, z_keyexpr("/a"), reply_handler, NULL, z_gopt_default());

Python

# --- SYNC
import zenoh.sync as zenoh

# Callback
session.get("/a", callback=reply_handler)
# Stream
let replies = session.get("/a")

# --- ASYNC
import zenoh.async as zenoh

# Callback
await session.get("/a", callback=reply_handler)
# Stream
let replies = await session.get("/a")