Following resources with WebSub

WebSub is a W3C protocol that lets a subscriber be told when a resource (a topic) changes, instead of polling it. Since MAMS 2.4.0, agents can publish their resources as WebSub topics and subscribe to topics on other MAMS services.

This page uses MAMS 2.4.1, ASTRA 2.6.1 and Java 17 or later. Complete Your first MAMS service and read Response headers and links first.

How it works

The publisher has a hub, a resource at <agent>/websub. Its topics advertise the hub in their Link header (rel="hub"), along with their own URI (rel="self").
A subscriber creates a callback resource and asks the hub to subscribe it to a topic. To verify the subscription, the hub asks the callback to echo a random challenge.
When the publisher publishes a topic, the hub sends the topic’s current content to each subscriber’s callback. The subscriber agent receives it as a !topicUpdate goal.

Both agents extend mams.WebSub, from mams-astra-websub:

<dependency>
  <groupId>mams-ucd</groupId>
  <artifactId>mams-astra-websub</artifactId>
  <version>2.4.1</version>
</dependency>

Use the mams-cartago repository and the astra-cartago-base:2.6.1 parent, as in Your first MAMS service.

A topic

A topic is a resource whose representation subscribers receive. This one returns its value as JSON (src/main/java/example/TopicArtifact.java):

package example;

import cartago.ARTIFACT_INFO;
import cartago.OPERATION;
import cartago.OUTPORT;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import mams.artifacts.ResourceArtifact;
import mams.handlers.ResourceHandler;
import mams.web.ResponseEntity;
import mams.web.WebServer;

@ARTIFACT_INFO(outports = { @OUTPORT(name = "out-1") })
public class TopicArtifact extends ResourceArtifact {
    private volatile int value;

    @OPERATION
    void init(String name) {
        this.name = name;
    }

    @OPERATION
    void set(int value) {
        this.value = value;
    }

    @Override
    protected ResourceHandler createHandler() {
        return new ResourceHandler(this);
    }

    @Override
    public boolean handle(ChannelHandlerContext ctx, FullHttpRequest request) {
        WebServer.writeResponse(ctx, request, ResponseEntity.type("application/json")
                .status(HttpResponseStatus.OK).body("{\"value\": " + value + "}"));
        return true;
    }
}

The publisher

The publisher creates its hub and the topic, and links the topic to the hub. Then it creates a subscriber, waits four seconds and publishes (src/main/astra/Publisher.astra):

agent Publisher extends mams.WebSub {
    module Timer timer;
    module System system;
    module Console console;

    types p { formula topic(string); }

    rule +!main(list args) {
        MAMSAgent::!setup(9000);
        MAMSAgent::!created("base");
        WebSub::!have("websub");
        MAMSAgent::!createResource("deployment", "base", "example.TopicArtifact",
            ["deployment"], cartago.ArtifactId topicId);
        query(uri(system.name() + "-base-deployment", string topic));
        +topic(topic);
        query(uri(system.name() + "-base-websub", string hub));
        WebSub::!linkToResource("deployment");
        cartago.operation(topicId, set(42));
        system.createAgent("subscriber", "Subscriber");
        system.setMainGoal("subscriber", [topic, hub]);
        timer.every("publish", 4000);
    }

    rule $timer.tick("publish") : topic(string topic) {
        timer.cancel("publish");
        WebSub::!publish(topic);
    }
}

!createResource records the topic’s URI as uri(ResName, Uri). The resource name is the agent name, the parent resource and the resource name, joined by -. !linkToResource adds the topic’s hub link. !publish sends the topic’s current representation to every subscriber; ?subscribers(topic, list urls) lists their callbacks.

The subscriber

The subscriber creates a callback for the hub, subscribes, and handles updates (src/main/astra/Subscriber.astra). In this example it runs in the same JVM, but it would normally be another service:

agent Subscriber extends mams.WebSub {
    module System system;
    module Console console;

    rule +!main([string topic, string hub]) {
        MAMSAgent::!init();
        MAMSAgent::!created("base");
        WebSub::!have("websub");
        WebSub::!registerHub("hub", hub);
        WebSub::!subscribe(topic, "hub");
    }

    rule +!topicUpdate(string self, string hub, string type, string content) {
        console.println("UPDATE " + self + " " + type + " " + content);
        system.exit();
    }
}

In the POM’s <properties>, set astra.main to Publisher and astra.name to publisher (the agent’s name, which is part of its resources’ URIs). Then run mvn compile astra:run. After four seconds the subscriber prints the topic’s URI, its media type and its content:

[subscriber] UPDATE http://<host>:9000/publisher/deployment application/json {"value": 42}

Reference

Goal or event	Meaning
`!have("websub")`	Create this agent’s WebSub resource (`<agent>/websub`): its hub, and the parent of its callbacks.
`!linkToResource(string Name)`	Advertise this agent’s hub from the resource `Name` (a `Link` with `rel="hub"`).
`!publish(string Topic)`	Fetch the topic and deliver its content to its subscribers.
`?subscribers(string Topic, list Callbacks)`	The callbacks subscribed to a topic.
`!registerHub(string Name, string HubUri)`	Create a callback resource (`<agent>/websub/<Name>`) for a hub.
`!subscribe(string Topic, string Name)`	Subscribe the callback `Name` to a topic.
`+!topicUpdate(string Self, string Hub, string Type, string Content)`	Implement this to receive updates: the topic’s `self` and `hub` links, media type and content.

The module guide describes the artifacts. A GET to a callback without hub.challenge is rejected with 400.