WebSockets Next 参考指南

import jakarta.enterprise.context.SessionScoped;

@WebSocket(path = "/ws")
@SessionScoped (1)
public class MyWebSocket {

}

import jakarta.enterprise.context.RequestScoped;

@WebSocket(path = "/ws")
@RequestScoped (1)
public class MyWebSocket {

}

// No need to subscribe to the incoming Multi as the method returns a Multi derived from the incoming one
@OnTextMessage
public Multi<ChatMessage> stream(Multi<ChatMessage> incoming) {
    return incoming.log();
}

// ...

// Must subscribe to the incoming Multi as the method does not return a Multi, otherwise no data will be consumed
@OnTextMessage
public void stream(Multi<ChatMessage> incoming) {
    incoming.subscribe().with(item -> log(item));
}

@OnTextMessage
void consume(Message m) {
// Process the incoming message. The method is called on an executor thread for each incoming message.
}

@OnTextMessage
Uni<Void> consumeAsync(Message m) {
// Process the incoming message. The method is called on an event loop thread for each incoming message.
// The method completes when the returned Uni emits its item.
}

@OnTextMessage
ResponseMessage process(Message m) {
// Process the incoming message and send a response to the client.
// The method is called for each incoming message.
// Note that if the method returns `null`, no response will be sent to the client.
}

@OnTextMessage
Uni<ResponseMessage> processAsync(Message m) {
// Process the incoming message and send a response to the client.
// The method is called for each incoming message.
// Note that if the method returns `null`, no response will be sent to the client. The method completes when the returned Uni emits its item.
}

@OnTextMessage
Multi<ResponseMessage> stream(Message m) {
// Process the incoming message and send multiple responses to the client.
// The method is called for each incoming message.
// The method completes when the returned Multi emits its completion signal.
// The method cannot return `null` (but an empty multi if no response must be sent)
}

@OnTextMessage
public Multi<ChatMessage> stream(Multi<ChatMessage> incoming) {
    return incoming.log();
}

@OnTextMessage
public void stream(Multi<ChatMessage> incoming) {
    incoming.subscribe().with(item -> log(item));
}

@OnOpen(broadcast = true)
public ChatMessage onOpen() {
    return new ChatMessage(MessageType.USER_JOINED, connection.pathParam("username"), null);
}

@Inject WebSocketConnection connection;

@OnClose
public void onClose() {
    ChatMessage departure = new ChatMessage(MessageType.USER_LEFT, connection.pathParam("username"), null);
    connection.broadcast().sendTextAndAwait(departure);
}

@Singleton
public class ItemBinaryMessageCodec implements BinaryMessageCodec<Item> {

    @Override
    public boolean supports(Type type) {
        // Allows selecting the right codec for the right type
        return type.equals(Item.class);
    }

    @Override
    public Buffer encode(Item value) {
        // Serialization
        return Buffer.buffer(value.toString());
    }

    @Override
    public Item decode(Type type, Buffer value) {
        // Deserialization
        return new Item(value.toString());
    }
}

@OnTextMessage(codec = MyInputCodec.class) (1)
Item find(Item item) {
        //....
}

@OnTextMessage(
        codec = MyInputCodec.class, (1)
        outputCodec = MyOutputCodec.class (2)
Item find(Item item) {
        //....
}

quarkus.websockets-next.server.auto-ping-interval=2 (1)
quarkus.websockets-next.client.auto-ping-interval=10 (2)

@OnPingMessage
void ping(Buffer data) {
    // an incoming ping that will automatically receive a pong
}

@OnPongMessage
void pong(Buffer data) {
    // an incoming pong in response to the last ping sent
}

import io.quarkus.logging.Log;
import io.quarkus.websockets.next.OpenConnections;

class MyBean {

  @Inject
  OpenConnections connections;

  void logAllOpenConnections() {
     Log.infof("Open connections: %s", connections.listAll()); (1)
  }
}

import io.quarkus.websockets.next.WebSocketConnection;
import io.quarkus.websockets.next.UserData.TypedKey;

@WebSocket(path = "/endpoint/{username}")
class MyEndpoint {

  @Inject
  CoolService service;

  @OnOpen
  void open(WebSocketConnection connection) {
     connection.userData().put(TypedKey.forBoolean("isCool"), service.isCool(connection.pathParam("username"))); (1)
  }

  @OnTextMessage
  String process(String message) {
     if (connection.userData().get(TypedKey.forBoolean("isCool"))) { (2)
        return "Cool message processed!";
     } else {
        return "Message processed!";
     }
  }
}

import jakarta.enterprise.event.ObservesAsync;
import io.quarkus.websockets.next.Open;
import io.quarkus.websockets.next.WebSocketConnection;

class MyBean {

  void connectionOpened(@ObservesAsync @Open WebSocketConnection connection) { (1)
     // This observer method is called when a connection is opened...
  }
}

package io.quarkus.websockets.next.test.security;

import io.quarkus.security.Authenticated;
import jakarta.inject.Inject;

import io.quarkus.security.identity.SecurityIdentity;
import io.quarkus.websockets.next.OnOpen;
import io.quarkus.websockets.next.OnTextMessage;
import io.quarkus.websockets.next.WebSocket;

@Authenticated (1)
@WebSocket(path = "/end")
public class Endpoint {

    @Inject
    SecurityIdentity currentIdentity;

    @OnOpen
    String open() {
        return "ready";
    }

    @OnTextMessage
    String echo(String message) {
        return message;
    }
}

quarkus.http.auth.permission.http-upgrade.paths=/end
quarkus.http.auth.permission.http-upgrade.policy=authenticated

package io.quarkus.websockets.next.test.security;

import io.quarkus.oidc.BearerTokenAuthentication;
import io.quarkus.websockets.next.OnTextMessage;
import io.quarkus.websockets.next.WebSocket;

@BearerTokenAuthentication (1)
@WebSocket(path = "/end")
public class Endpoint {

    @OnTextMessage
    String echo(String message) {
        return message;
    }

}

quarkus.http.auth.proactive=false (1)

package io.quarkus.websockets.next.test.security;

import jakarta.annotation.security.RolesAllowed;
import jakarta.inject.Inject;

import io.quarkus.security.ForbiddenException;
import io.quarkus.security.identity.SecurityIdentity;
import io.quarkus.websockets.next.OnError;
import io.quarkus.websockets.next.OnOpen;
import io.quarkus.websockets.next.OnTextMessage;
import io.quarkus.websockets.next.WebSocket;

@WebSocket(path = "/end")
public class Endpoint {

    @Inject
    SecurityIdentity currentIdentity;

    @OnOpen
    String open() {
        return "ready";
    }

    @RolesAllowed("admin")
    @OnTextMessage
    String echo(String message) { (1)
        return message;
    }

    @OnError
    String error(ForbiddenException t) { (2)
        return "forbidden:" + currentIdentity.getPrincipal().getName();
    }
}

package io.quarkus.websockets.next.test.security;

import io.quarkus.security.PermissionsAllowed;
import io.quarkus.websockets.next.OnTextMessage;
import io.quarkus.websockets.next.WebSocket;

@PermissionsAllowed("product:premium")
@WebSocket(path = "/product/premium")
public class PremiumProductEndpoint {

    @OnTextMessage
    PremiumProduct getPremiumProduct(int productId) {
        return new PremiumProduct(productId);
    }

}

package io.quarkus.websockets.next.test.security;

import io.quarkus.security.identity.SecurityIdentity;
import io.quarkus.security.PermissionChecker;
import io.quarkus.vertx.http.runtime.security.HttpSecurityUtils;
import io.vertx.ext.web.RoutingContext;
import jakarta.enterprise.context.ApplicationScoped;

@ApplicationScoped
public class PermissionChecker {

    @PermissionChecker("product:premium")
    public boolean canGetPremiumProduct(SecurityIdentity securityIdentity) { (1)
        String username = securityIdentity.getPrincipal().getName();

        RoutingContext routingContext = HttpSecurityUtils.getRoutingContextAttribute(securityIdentity);
        String initialHttpUpgradePath = routingContext == null ? null : routingContext.normalizedPath();
        if (!isUserAllowedToAccessPath(initialHttpUpgradePath, username)) {
            return false;
        }

        return isPremiumCustomer(username);
    }

}

package io.quarkus.websockets.next.test.security;

import io.quarkus.security.PermissionChecker;
import io.quarkus.security.PermissionsAllowed;
import jakarta.inject.Inject;

import io.quarkus.security.ForbiddenException;
import io.quarkus.security.identity.SecurityIdentity;
import io.quarkus.websockets.next.OnError;
import io.quarkus.websockets.next.OnOpen;
import io.quarkus.websockets.next.OnTextMessage;
import io.quarkus.websockets.next.WebSocket;

@WebSocket(path = "/product")
public class ProductEndpoint {

    private record Product(int id, String name) {}

    @Inject
    SecurityIdentity currentIdentity;

    @PermissionsAllowed("product:get")
    @OnTextMessage
    Product getProduct(int productId) { (1)
        return new Product(productId, "Product " + productId);
    }

    @OnError
    String error(ForbiddenException t) { (2)
        return "forbidden:" + currentIdentity.getPrincipal().getName();
    }

    @PermissionChecker("product:get")
    boolean canGetProduct(int productId) {
        String username = currentIdentity.getPrincipal().getName();
        return currentIdentity.hasRole("admin") || canUserGetProduct(productId, username);
    }
}

const token = getBearerToken()
const quarkusHeaderProtocol = encodeURIComponent("quarkus-http-upgrade#Authorization#Bearer " + token) (1)
const socket = new WebSocket("wss://" + location.host + "/chat/" + username, ["bearer-token-carrier", quarkusHeaderProtocol]) (2)

quarkus.websockets-next.server.supported-subprotocols=bearer-token-carrier
quarkus.websockets-next.server.propagate-subprotocol-headers=true

@Singleton
public class MyBean {

    @Inject
    BasicWebSocketConnector connector; (1)

    void openAndConsume() {
        WebSocketClientConnection connection = connector
            .baseUri(uri) (2)
            .path("/ws") (3)
            .executionModel(ExecutionModel.NON_BLOCKING) (4)
            .onTextMessage((c, m) -> { (5)
               // ...
            })
            .connectAndAwait();
    }
}

import jakarta.enterprise.inject.Instance;

@Singleton
public class MyBean {

    @Inject
    Instance<BasicWebSocketConnector> connector;

    void connect() {
        var connection1 = connector.get().baseUri(uri)
                .addHeader("Foo", "alpha")
                .connectAndAwait();
        var connection2 = connector.get().baseUri(uri)
                .addHeader("Foo", "bravo")
                .connectAndAwait();
    }
}

import io.quarkus.logging.Log;
import io.quarkus.websockets.next.OpenClientConnections;

class MyBean {

  @Inject
  OpenClientConnections connections;

  void logAllOpenClinetConnections() {
     Log.infof("Open client connections: %s", connections.listAll()); (1)
  }
}

import io.quarkus.websockets.next.WebSocketClientConnection;
import io.quarkus.websockets.next.UserData.TypedKey;

@WebSocketClient(path = "/endpoint/{username}")
class MyEndpoint {

  @Inject
  CoolService service;

  @OnOpen
  void open(WebSocketClientConnection connection) {
     connection.userData().put(TypedKey.forBoolean("isCool"), service.isCool(connection.pathParam("username"))); (1)
  }

  @OnTextMessage
  String process(String message) {
     if (connection.userData().get(TypedKey.forBoolean("isCool"))) { (2)
        return "Cool message processed!";
     } else {
        return "Message processed!";
     }
  }
}

import jakarta.enterprise.event.ObservesAsync;
import io.quarkus.websockets.next.Open;
import io.quarkus.websockets.next.WebSocketClientConnection;

class MyBean {

  void connectionOpened(@ObservesAsync @Open WebSocketClientConnection connection) { (1)
     // This observer method is called when a connection is opened...
  }
}