package io.vertx.ext.stomp;
import io.vertx.core.Handler;
import io.vertx.ext.stomp.utils.Headers;
import java.util.List;
public class DefaultSubscribeHandler implements Handler<ServerFrame> {
@Override
public void handle(ServerFrame serverFrame) {
Frame frame = serverFrame.frame();
StompServerConnection connection = serverFrame.connection();
String id = frame.getHeader(Frame.ID);
String destination = frame.getHeader(Frame.DESTINATION);
String ack = frame.getHeader(Frame.ACK);
if (ack == null) {
ack = "auto";
}
if (destination == null || id == null) {
connection.write(Frames.createErrorFrame(
"Invalid subscription",
Headers.create(
frame.getHeaders()), "The 'destination' and 'id' headers must be set"));
connection.close();
return;
}
int count = 0;
for (Destination dest : connection.handler().getDestinations()) {
List<String> ids = dest.getSubscriptions(connection);
count += ids.size();
if (ids.contains(id)) {
connection.write(Frames.createErrorFrame(
"Invalid subscription",
Headers.create(frame.getHeaders()), "'id'" +
" already used by this connection."));
connection.close();
return;
}
if (count + 1 > connection.server().options().getMaxSubscriptionsByClient()) {
connection.write(Frames.createErrorFrame(
"Invalid subscription",
Headers.create(frame.getHeaders()), "Too many subscriptions"));
connection.close();
return;
}
}
final Destination dest = connection.handler().getOrCreateDestination(destination);
if (dest != null) {
if (dest.subscribe(connection, frame) == null) {
connection.write(Frames.createErrorFrame(
"Access denied",
Headers.create(frame.getHeaders()), "The destination has been rejected by the server"));
connection.close();
return;
}
} else {
connection.write(Frames.createErrorFrame(
"Invalid subscription",
Headers.create(frame.getHeaders()), "The destination has been rejected by the server"));
connection.close();
return;
}
Frames.handleReceipt(frame, connection);
}
}