13 CoAP Observe: Subscriptions and Updates

13.8.1 Traditional Polling vs. Observe

The energy question is not simply “push or poll”; it is how many exchanges occur when a resource changes less often than a client would ask for it. The timelines in Figure 13.1 show the exchanges generated when client requests outpace changes to a resource.

+

Read Figure 13.1 from left to right along each lane. Polling repeats a GET and response even when the value has not changed; Observe pays once to register, then the server emits notifications when the representation changes. The diagram explains the message-count difference, while the figures below quantify it for one assumed polling interval and change rate rather than claiming the same saving for every workload.

Traffic comparison for 24 hours of temperature monitoring:

• Polling (every 60 seconds): 1,440 requests + 1,440 responses, about 61.9 KiB, high battery impact.
• Observe (10 changes/hour): 1 registration + 240 notifications, about 6.6 KiB, about 89% less traffic.

13.8.2 Observer Architecture Overview

Before treating notifications as isolated responses, inspect Figure 13.2 to see the state that the client, wire exchange, and server must preserve together.

+

Start with one client. Track its own saved state. Read the architecture in Figure 13.2 from client state through Token and Observe ordering on the wire to the server’s observer registry. Then follow failure ownership and the bounded evidence required before the application accepts a notification. Multiple clients can observe one resource, but each relationship needs its own retained state; the next sections trace registration and notification behavior inside that contract.

+Checkpoint: Push or Poll?

• You now know why polling every 60 seconds creates 1,440 requests and 1,440 responses over 24 hours.
• You now know how Observe reduces that same example to one registration plus 240 notifications, about 89% less traffic.
• You now know the first design question: whether the resource changes far less often than the client would poll it.

13.9.1 Registration

Client sends GET with Observe: 0 to register:

Request

• Client -> Server: GET /temperature
• Token: 0xAB12
• Observe: 0 (register)
• Accept: text/plain

Response

• Server -> Client: 2.05 Content
• Token: 0xAB12
• Observe: 1 (sequence number)
• Max-Age: 60
• Payload: "23.5"

13.9.2 Notifications

Inspect Figure 13.3 to see how a client proves that each notification belongs to its observation and is fresher than the value already accepted.

+

Watch one subscription from start to end. Track its state. Trace Figure 13.3 from registration through successive notifications. Token continuity identifies the observation, while Observe values and freshness rules help reject stale or reordered updates; a transport ACK confirms delivery but does not by itself prove that the application accepted the value. Cancellation or relationship loss ends that state, and re-registration requires reconciliation. That lifecycle frames the sequence-number knowledge check next.

13.9.3 Deregistration

Three ways to stop receiving notifications:

1. Explicit deregistration (GET with Observe: 1):

• Client -> Server: GET /temperature
• Token: 0xAB12
• Observe: 1 (deregister)

2. RST response to unwanted notification:

• Server -> Client: NON 2.05 Content (notification)
• Client -> Server: RST (stop this observation)

3. Timeout (Max-Age expiration): Keep one practical point in view: If the client does not refresh the observation within Max-Age,.

• If the client does not refresh the observation within Max-Age, the server removes that observer entry.

The protocol flow gives the client-side contract: register with Observe: 0, accept ordered notifications, and leave cleanly with Observe: 1 or RST. The next layer is the server-side contract that remembers each observer without letting stale entries accumulate.

13.10.1 Server-Side Observer Registry

from collections import defaultdict
from dataclasses import dataclass
import time

@dataclass
class Observer:
    client_addr: object
    token: bytes
    request_options: dict  # Effective GET options that select/cache the representation.
    registered_at: float
    last_notification: float
    timeout: float

class ObserverRegistry:
    def __init__(self):
        # Map: resource_uri -> list of Observer objects
        self.observers = defaultdict(list)
        self.observer_timeout = 86400  # 24 hours default

    def register_observer(self, resource_uri, client_addr, token, request_options,
                          lease_seconds=None):
        # Include effective Accept and other representation-selecting GET options.
        options_snapshot = dict(request_options)
        # RFC 7641 §4.1: refresh an existing endpoint/token pair, do not append it twice.
        for existing in self.observers[resource_uri]:
            if existing.client_addr == client_addr and existing.token == token:
                existing.registered_at = time.time()
                existing.timeout = lease_seconds or self.observer_timeout
                existing.request_options = options_snapshot
                return existing
        observer = Observer(
            client_addr=client_addr,
            token=token,
            request_options=options_snapshot,
            registered_at=time.time(),
            last_notification=time.time(),
            timeout=lease_seconds or self.observer_timeout  # application policy, not Max-Age
        )
        self.observers[resource_uri].append(observer)
        return observer

    def notify_all(self, resource_uri):
        """Send notification to all observers of a resource"""
        expired = []

        for observer in self.observers[resource_uri]:
            # Check if observation expired
            if time.time() - observer.registered_at > observer.timeout:
                expired.append(observer)
                continue

            # Re-evaluate this observer's selected representation using its GET options.
            value, content_format = self.select_representation(
                resource_uri, observer.request_options
            )
            self.send_notification(observer, value, content_format)

        # Clean up expired observers
        for observer in expired:
            self.observers[resource_uri].remove(observer)

    def remove_observer(self, resource_uri, client_addr, token):
        """Remove observer on explicit deregistration or RST received"""
        self.observers[resource_uri] = [
            o for o in self.observers[resource_uri]
            if not (o.client_addr == client_addr and o.token == token)
        ]

13.10.2 Notification Rate Limiting

Prevent notification floods on rapidly-changing resources:

class RateLimitedResource:
    def __init__(self, min_interval=1.0, change_threshold=0.5,
                 urgent_alarm_threshold=None):
        self.min_interval = min_interval    # Minimum seconds between notifications
        self.change_threshold = change_threshold  # Minimum change to trigger notification
        self.urgent_alarm_threshold = urgent_alarm_threshold
        self.last_notify_time = {}          # Per-observer last notification time
        self.last_notified_value = {}       # Per-observer last sent value

    def on_value_change(self, new_value):
        now = time.time()

        for observer in self.observers:
            last_time = self.last_notify_time.get(observer.token, 0)
            last_value = self.last_notified_value.get(observer.token, None)

            # Routine readings obey both gates; an alarm crossing bypasses the time gate.
            routine_notify = (
                last_value is None or
                (abs(new_value - last_value) >= self.change_threshold and
                 (now - last_time) >= self.min_interval)
            )
            urgent_alarm = (
                self.urgent_alarm_threshold is not None and
                new_value >= self.urgent_alarm_threshold and
                (last_value is None or last_value < self.urgent_alarm_threshold)
            )

            if routine_notify or urgent_alarm:
                self.send_notification(observer, new_value)
                self.last_notify_time[observer.token] = now
                self.last_notified_value[observer.token] = new_value

13.12.1 Token Reuse After Client Restart

Problem:

- Client registers observation with Token=0x42
- Client crashes and restarts
- Server sends notification with Token=0x42
- Client doesn't recognize token (state lost) -> sends RST
- Server removes observer

Solutions:

1. Server MUST remove observer when RST received
2. Client should re-register observations after restart
3. Consider persisting observation state to flash

13.12.2 NAT Timeout Issue

Problem:

UDP NAT mapping lifetime varies by network; this example assumes 45 seconds
- Client behind NAT registers observation
- Server tries to push notification 5 minutes later
- NAT mapping expired -> notification never reaches client

Solutions:

1. Client sends periodic traffic before its measured NAT mapping expires
2. Client sends periodic re-registration (GET with Observe=0)
3. Use Max-Age to bound representation freshness; plan notifications or re-registration accordingly
4. Consider CoAP over TCP for NAT-hostile networks