javax.ws.rs.core.Response.Status#METHOD_NOT_ALLOWED源码实例Demo

public void locate(String microserviceName, String path, String httpMethod, MicroservicePaths microservicePaths) {
  // 在静态路径中查找
  operation = locateStaticPathOperation(path, httpMethod, microservicePaths.getStaticPathOperationMap());
  if (operation != null) {
    // 全部定位完成
    return;
  }

  // 在动态路径中查找
  operation = locateDynamicPathOperation(path, microservicePaths.getDynamicPathOperationList(), httpMethod);
  if (operation != null) {
    return;
  }

  Status status = Status.NOT_FOUND;
  if (resourceFound) {
    status = Status.METHOD_NOT_ALLOWED;
  }
  LOGGER.error("locate path failed, status:{}, http method:{}, path:{}, microserviceName:{}",
      status,
      httpMethod,
      path,
      microserviceName);
  throw new InvocationException(status, status.getReasonPhrase());
}
 

protected MessageId internalTerminate(boolean authoritative) {
    if (topicName.isGlobal()) {
        validateGlobalNamespaceOwnership(namespaceName);
    }
    PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false);
    if (partitionMetadata.partitions > 0) {
        throw new RestException(Status.METHOD_NOT_ALLOWED, "Termination of a partitioned topic is not allowed");
    }
    validateWriteOperationOnTopic(authoritative);
    Topic topic = getTopicReference(topicName);
    try {
        return ((PersistentTopic) topic).terminate().get();
    } catch (Exception exception) {
        log.error("[{}] Failed to terminated topic {}", clientAppId(), topicName, exception);
        throw new RestException(exception);
    }
}
 

protected void internalSkipMessages(String subName, int numMessages, boolean authoritative) {
    if (topicName.isGlobal()) {
        validateGlobalNamespaceOwnership(namespaceName);
    }
    PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false);
    if (partitionMetadata.partitions > 0) {
        throw new RestException(Status.METHOD_NOT_ALLOWED, "Skip messages on a partitioned topic is not allowed");
    }
    validateAdminAccessForSubscriber(subName, authoritative);
    PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
    try {
        if (subName.startsWith(topic.getReplicatorPrefix())) {
            String remoteCluster = PersistentReplicator.getRemoteCluster(subName);
            PersistentReplicator repl = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster);
            Preconditions.checkNotNull(repl);
            repl.skipMessages(numMessages).get();
        } else {
            PersistentSubscription sub = topic.getSubscription(subName);
            Preconditions.checkNotNull(sub);
            sub.skipMessages(numMessages).get();
        }
        log.info("[{}] Skipped {} messages on {} {}", clientAppId(), numMessages, topicName, subName);
    } catch (NullPointerException npe) {
        throw new RestException(Status.NOT_FOUND, "Subscription not found");
    } catch (Exception exception) {
        log.error("[{}] Failed to skip {} messages {} {}", clientAppId(), numMessages, topicName, subName,
                exception);
        throw new RestException(exception);
    }
}
 

private void verifyReadOperation(boolean authoritative) {
    if (topicName.isGlobal()) {
        validateGlobalNamespaceOwnership(namespaceName);
    }
    PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(topicName, authoritative, false);
    if (partitionMetadata.partitions > 0) {
        throw new RestException(Status.METHOD_NOT_ALLOWED, "Peek messages on a partitioned topic is not allowed");
    }
}
 

private void validateClientVersion() {
    if (!pulsar().getConfiguration().isClientLibraryVersionCheckEnabled()) {
        return;
    }
    final String userAgent = httpRequest.getHeader("User-Agent");
    if (StringUtils.isBlank(userAgent)) {
        throw new RestException(Status.METHOD_NOT_ALLOWED,
                "Client lib is not compatible to access partitioned metadata: version in user-agent is not present");
    }
    // Version < 1.20 for cpp-client is not allowed
    if (userAgent.contains(DEPRECATED_CLIENT_VERSION_PREFIX)) {
        try {
            // Version < 1.20 for cpp-client is not allowed
            String[] tokens = userAgent.split(DEPRECATED_CLIENT_VERSION_PREFIX);
            String[] splits = tokens.length > 1 ? tokens[1].split("-")[0].trim().split("\\.") : null;
            if (splits != null && splits.length > 1) {
                if (LEAST_SUPPORTED_CLIENT_VERSION_PREFIX.getMajorVersion() > Integer.parseInt(splits[0])
                        || LEAST_SUPPORTED_CLIENT_VERSION_PREFIX.getMinorVersion() > Integer.parseInt(splits[1])) {
                    throw new RestException(Status.METHOD_NOT_ALLOWED,
                            "Client lib is not compatible to access partitioned metadata: version " + userAgent
                                    + " is not supported");
                }
            }
        } catch (RestException re) {
            throw re;
        } catch (Exception e) {
            log.warn("[{}] Failed to parse version {} ", clientAppId(), userAgent);
        }
    }
    return;
}
 

protected void internalResetCursorOnPosition(String subName, boolean authoritative, MessageIdImpl messageId) {
    if (topicName.isGlobal()) {
        validateGlobalNamespaceOwnership(namespaceName);
    }
    log.info("[{}][{}] received reset cursor on subscription {} to position {}", clientAppId(), topicName,
            subName, messageId);
    // If the topic name is a partition name, no need to get partition topic metadata again
    if (!topicName.isPartitioned() && getPartitionedTopicMetadata(topicName, authoritative, false).partitions > 0) {
        log.warn("[{}] Not supported operation on partitioned-topic {} {}", clientAppId(), topicName,
                subName);
        throw new RestException(Status.METHOD_NOT_ALLOWED,
                "Reset-cursor at position is not allowed for partitioned-topic");
    } else {
        validateAdminAccessForSubscriber(subName, authoritative);
        PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
        if (topic == null) {
            throw new RestException(Status.NOT_FOUND, "Topic not found");
        }
        try {
            PersistentSubscription sub = topic.getSubscription(subName);
            Preconditions.checkNotNull(sub);
            sub.resetCursor(PositionImpl.get(messageId.getLedgerId(), messageId.getEntryId())).get();
            log.info("[{}][{}] successfully reset cursor on subscription {} to position {}", clientAppId(),
                    topicName, subName, messageId);
        } catch (Exception e) {
            Throwable t = e.getCause();
            log.warn("[{}] [{}] Failed to reset cursor on subscription {} to position {}", clientAppId(),
                    topicName, subName, messageId, e);
            if (e instanceof NullPointerException) {
                throw new RestException(Status.NOT_FOUND, "Subscription not found");
            } else if (t instanceof SubscriptionInvalidCursorPosition) {
                throw new RestException(Status.PRECONDITION_FAILED,
                        "Unable to find position for position specified: " + t.getMessage());
            } else if (t instanceof SubscriptionBusyException) {
                throw new RestException(Status.PRECONDITION_FAILED,
                        "Failed for SubscriptionBusy: " + t.getMessage());
            } else {
                throw new RestException(e);
            }
        }
    }
}
 

protected Response internalPeekNthMessage(String subName, int messagePosition, boolean authoritative) {
    verifyReadOperation(authoritative);
    // If the topic name is a partition name, no need to get partition topic metadata again
    if (!topicName.isPartitioned() && getPartitionedTopicMetadata(topicName, authoritative, false).partitions > 0) {
        throw new RestException(Status.METHOD_NOT_ALLOWED, "Peek messages on a partitioned topic is not allowed");
    }
    validateAdminAccessForSubscriber(subName, authoritative);
    if (!(getTopicReference(topicName) instanceof PersistentTopic)) {
        log.error("[{}] Not supported operation of non-persistent topic {} {}", clientAppId(), topicName,
                subName);
        throw new RestException(Status.METHOD_NOT_ALLOWED,
                "Skip messages on a non-persistent topic is not allowed");
    }

    PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
    PersistentReplicator repl = null;
    PersistentSubscription sub = null;
    Entry entry = null;
    if (subName.startsWith(topic.getReplicatorPrefix())) {
        repl = getReplicatorReference(subName, topic);
    } else {
        sub = (PersistentSubscription) getSubscriptionReference(subName, topic);
    }
    try {
        if (subName.startsWith(topic.getReplicatorPrefix())) {
            entry = repl.peekNthMessage(messagePosition).get();
        } else {
            entry = sub.peekNthMessage(messagePosition).get();
        }
        return generateResponseWithEntry(entry);
    } catch (NullPointerException npe) {
        throw new RestException(Status.NOT_FOUND, "Message not found");
    } catch (Exception exception) {
        log.error("[{}] Failed to get message at position {} from {} {}", clientAppId(), messagePosition,
                topicName, subName, exception);
        throw new RestException(exception);
    } finally {
        if (entry != null) {
            entry.release();
        }
    }
}
 

private void internalExpireMessagesForSinglePartition(String subName, int expireTimeInSeconds,
        boolean authoritative) {
    if (topicName.isGlobal()) {
        validateGlobalNamespaceOwnership(namespaceName);
    }
    // If the topic name is a partition name, no need to get partition topic metadata again
    if (!topicName.isPartitioned() && getPartitionedTopicMetadata(topicName, authoritative, false).partitions > 0) {
        String msg = "This method should not be called for partitioned topic";
        log.error("[{}] {} {} {}", clientAppId(), msg, topicName, subName);
        throw new IllegalStateException(msg);
    }

    // validate ownership and redirect if current broker is not owner
    validateAdminAccessForSubscriber(subName, authoritative);

    if (!(getTopicReference(topicName) instanceof PersistentTopic)) {
        log.error("[{}] Not supported operation of non-persistent topic {} {}", clientAppId(), topicName, subName);
        throw new RestException(Status.METHOD_NOT_ALLOWED,
                "Expire messages on a non-persistent topic is not allowed");
    }

    PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
    try {
        if (subName.startsWith(topic.getReplicatorPrefix())) {
            String remoteCluster = PersistentReplicator.getRemoteCluster(subName);
            PersistentReplicator repl = (PersistentReplicator) topic.getPersistentReplicator(remoteCluster);
            Preconditions.checkNotNull(repl);
            repl.expireMessages(expireTimeInSeconds);
        } else {
            PersistentSubscription sub = topic.getSubscription(subName);
            Preconditions.checkNotNull(sub);
            sub.expireMessages(expireTimeInSeconds);
        }
        log.info("[{}] Message expire started up to {} on {} {}", clientAppId(), expireTimeInSeconds, topicName,
                subName);
    } catch (NullPointerException npe) {
        throw new RestException(Status.NOT_FOUND, "Subscription not found");
    } catch (Exception exception) {
        log.error("[{}] Failed to expire messages up to {} on {} with subscription {} {}", clientAppId(),
                expireTimeInSeconds, topicName, subName, exception);
        throw new RestException(exception);
    }
}