Tutorial: Receiving a Session
This tutorial shows how to receive an incoming session, read messages from it, and send replies. This is the counterpart to Creating a Session, which covers the initiating side.
Prerequisites
- Completed Creating an App — you need the
appandconn_idobjects
Listen for an Incoming Session
The receiving side does not call create_session. Instead it calls listen_for_session_async, which blocks until the remote peer initiates a session (either a point-to-point connection or a group invitation).
use slim_session::Notification;
// The notification receiver rx comes from service.create_app(...)
// Wait for an incoming session invitation
while let Some(Ok(notification)) = rx.recv().await {
if let Notification::NewSession(ctx) = notification {
println!("Session received");
// ctx is used in the Receive Messages step below
break;
}
}
import asyncio
import slim_bindings
# Wait indefinitely for an incoming session
session = await local_app.listen_for_session_async(None)
print(f"Session received from: {session.source()}")
import (
"fmt"
"log"
slim "github.com/agntcy/slim-bindings-go/v2"
)
// Wait indefinitely for an incoming session (nil = no timeout)
session, err := app.ListenForSessionAsync(nil)
if err != nil {
log.Fatal(err)
}
fmt.Println("Session received")
// Wait indefinitely for an incoming session (null = no timeout)
Session session = app.listenForSession(null);
System.out.println("Session received");
// Wait indefinitely for an incoming session (null = no timeout)
val session = app.listenForSessionAsync(null)
println("Session received")
// Wait up to 60 seconds for an incoming session (pass null to wait forever)
const session = await app.listenForSessionAsync(60000);
console.log("Session received");
// Wait up to 60 seconds for an incoming session
using var session = await app.ListenForSessionAsync(TimeSpan.FromSeconds(60));
Console.WriteLine($"Session received: {session.Destination}");
// Wait up to 60 seconds for an incoming session
const session = await app.listenForSessionAsync(60000);
console.log("Session received");
Receive Messages
Once the session is established, call get_message_async in a loop to receive messages. The call blocks until a message arrives or the timeout expires.
// Spawn a receiver task on the session context
ctx.spawn_receiver(|mut msg_rx, _| async move {
loop {
match msg_rx.recv().await {
Some(Ok(msg)) => {
println!("Received: {}", String::from_utf8_lossy(msg.payload()));
}
Some(Err(e)) => {
eprintln!("Session error: {e}");
break;
}
None => break, // Session closed
}
}
});
import datetime
while True:
try:
msg = await session.get_message_async(
timeout=datetime.timedelta(seconds=30)
)
print("Received:", msg.payload.decode())
except Exception as e:
if "session closed" in str(e).lower():
break
continue
import (
"fmt"
"time"
)
for {
timeout := 30 * time.Second
msg, err := session.GetMessageAsync(&timeout)
if err != nil {
break
}
fmt.Println("Received:", string(msg.Payload))
}
import java.time.Duration;
while (true) {
try {
ReceivedMessage msg = session.getMessage(Duration.ofSeconds(30));
System.out.println("Received: " + new String(msg.payload()));
} catch (Exception e) {
break;
}
}
import java.time.Duration
while (true) {
try {
val msg = session.getMessageAsync(Duration.ofSeconds(30))
println("Received: " + String(msg.payload))
} catch (e: Exception) {
break
}
}
while (true) {
try {
const msg = await session.getMessageAsync(30000);
console.log("Received:", Buffer.from(msg.payload).toString());
} catch (e) {
break; // Session closed or timeout
}
}
while (true)
{
try
{
var msg = await session.GetMessageAsync(TimeSpan.FromSeconds(30));
Console.WriteLine($"Received: {msg.Text}");
}
catch (Exception ex) when (ex.Message.Contains("timeout"))
{
continue;
}
catch
{
break; // Session closed
}
}
while (true) {
try {
const msg = await session.getMessageAsync(30000);
const text = String.fromCharCode(...new Uint8Array(msg.payload));
console.log("Received:", text);
} catch (e) {
break; // Session closed or timeout
}
}
Reply to Messages
There are two ways to reply:
- Broadcast (
publish_and_wait_async/PublishAndWaitAsync) — sends to all current session participants. For point-to-point sessions this is just the remote peer; for group sessions every member receives it. - Direct reply (
publish_to_and_wait_async/PublishToAndWaitAsync/ReplyAsync) — uses the context from the received message to send back only to the original sender. Other group participants do not see the reply.
let session = ctx.session_arc().unwrap();
// Broadcast to all participants
session.publish(&channel_name, b"hello everyone".to_vec(), None, None).await?;
// Reply only to the sender (using the source from a received message)
session.publish_to(msg.source(), b"hello back".to_vec(), None, None).await?;
# Broadcast to all participants
await session.publish_and_wait_async(b"hello everyone", None, None)
# Reply only to the sender
await session.publish_to_and_wait_async(
msg.context,
b"hello back",
None, # payload_type
{}, # metadata
)
// Broadcast to all participants
if err := session.PublishAndWaitAsync([]byte("hello everyone"), nil, nil); err != nil {
log.Fatal(err)
}
// Reply only to the sender
if err := session.PublishToAndWaitAsync(msg.Context, []byte("hello back"), nil, nil); err != nil {
log.Fatal(err)
}
// Broadcast to all participants
session.publishAndWait("hello everyone".getBytes(), null, null);
// Reply only to the sender
session.publishToAndWait(msg.context(), "hello back".getBytes(), null, null);
// Broadcast to all participants
val broadcastHandle = session.publishAsync("hello everyone".toByteArray(), null, null)
broadcastHandle.waitAsync()
// Reply only to the sender
val replyHandle = session.publishToAsync(msg.context, "hello back".toByteArray(), null, null)
replyHandle.waitAsync()
// Broadcast to all participants
await session.publishAndWaitAsync(Buffer.from("hello everyone"), undefined, undefined);
// Reply only to the sender
await session.publishToAndWaitAsync(
msg.context,
Buffer.from("hello back"),
undefined,
undefined
);
// Broadcast to all participants
await session.PublishAsync("hello everyone");
// Reply only to the sender
await session.ReplyAsync(msg, "hello back");
// Broadcast to all participants
const broadcast = new Uint8Array("hello everyone".split('').map(c => c.charCodeAt(0)));
await session.publishAndWaitAsync(broadcast, undefined, undefined);
// Reply only to the sender
const reply = new Uint8Array("hello back".split('').map(c => c.charCodeAt(0)));
await session.publishToAndWaitAsync(msg.context, reply, undefined, undefined);
Close the Session
When the message loop exits — either because you are done or because the remote side closed the session — call close to release local resources and signal the remote peer that you are leaving.
For group sessions there are two close modes: CloseMode::Soft goes offline temporarily (you stay on the roster and can rejoin later, including after a process restart when using persistence); CloseMode::Hard terminates the session permanently. The examples below use a hard close. If you need to pause and resume, see Session Persistence.
// Session closed is signalled by None on the receiver; close on your side too
session.close().await?.await?;
await session.close_and_wait_async()
if err := session.CloseAndWaitAsync(); err != nil {
log.Fatal(err)
}
session.closeAndWait();
session.closeAndWaitAsync()
await session.closeAndWaitAsync();
await session.CloseAndWaitAsync();
await session.closeAndWaitAsync();