sequence
This paper mainly studies the MessagingAccessPoint of OpenMessaging
MessagingAccessPoint
openmessaging-java/openmessaging-api/src/main/java/io/openmessaging/MessagingAccessPoint.java
/**
* An instance of {@code MessagingAccessPoint} may be obtained from {@link OMS}, which is capable of creating {@code
* Producer}, {@code Consumer}, {@code ResourceManager}, and other facility entities.
* <p>
* For example:
* <pre>
* MessagingAccessPoint messagingAccessPoint = OMS.getMessagingAccessPoint("oms:rocketmq://[email protected]/us-east:default_space");
* messagingAccessPoint.startup();
* Producer producer = messagingAccessPoint.createProducer();
* producer.startup();
* producer.send(producer.createBytesMessage("HELLO_QUEUE"."HELLO_BODY".getBytes(Charset.forName("UTF-8"))));
* </pre>
*
* @version OMS 1.0.0
* @since OMS 1.0.0
*/
public interface MessagingAccessPoint extends ServiceLifecycle {
/**
* Returns the target OMS specification version of the specified vendor implementation.
*
* @return the OMS version of implementation
* @see OMS#specVersion
*/
String implVersion();
/**
* Returns the attributes of this {@code MessagingAccessPoint} instance.
* <p>
* There are some standard attributes defined by OMS for {@code MessagingAccessPoint}:
* <ul>
* <li> {@link OMSBuiltinKeys#ACCESS_POINTS}, the specified access points.
* <li> {@link OMSBuiltinKeys#DRIVER_IMPL}, the fully qualified class name of the specified MessagingAccessPoint's
* implementation, the default value is {@literal io.openmessaging.<driver_type>.MessagingAccessPointImpl}.
* <li> {@link OMSBuiltinKeys#REGION}, the region the resources reside in.
* <li> {@link OMSBuiltinKeys#ACCOUNT_ID}, the ID of the specific account system that owns the resource.
* </ul>
*
* @return the attributes
*/
KeyValue attributes();
/**
* Creates a new {@code Producer} for the specified {@code MessagingAccessPoint}.
*
* @return the created {@code Producer}
* @throws OMSRuntimeException ifthe {@code MessagingAccessPoint} fails to handle this request * due to some internal error */ Producer createProducer(); /** * Creates a new {@code Producer}for the specified {@code MessagingAccessPoint}
* with some preset attributes.
*
* @param attributes the preset attributes
* @return the created {@code Producer}
* @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request
* due to some internal error
*/
Producer createProducer(KeyValue attributes);
/**
* Creates a new {@code PushConsumer} for the specified {@code MessagingAccessPoint}.
* The returned {@code PushConsumer} isn't attached to any queue, * uses {@link PushConsumer#attachQueue(String, MessageListener)} to attach queues. * * @return the created {@code PushConsumer} * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request * due to some internal error */ PushConsumer createPushConsumer(); /** * Creates a new {@code PushConsumer} for the specified {@code MessagingAccessPoint} with some preset attributes. * * @param attributes the preset attributes * @return the created {@code PushConsumer} * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request * due to some internal error */ PushConsumer createPushConsumer(KeyValue attributes); /** * Creates a new {@code PullConsumer} for the specified {@code MessagingAccessPoint}. * * @return the created {@code PullConsumer} * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request * due to some internal error */ PullConsumer createPullConsumer(); /** * Creates a new {@code PullConsumer} for the specified {@code MessagingAccessPoint} with some preset attributes. * * @param attributes the preset attributes * @return the created {@code PullConsumer} * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request * due to some internal error */ PullConsumer createPullConsumer(KeyValue attributes); /** * Creates a new {@code StreamingConsumer} for the specified {@code MessagingAccessPoint}. * * @return the created {@code Stream} * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request * due to some internal error */ StreamingConsumer createStreamingConsumer(); /** * Creates a new {@code StreamingConsumer} for the specified {@code MessagingAccessPoint} with some preset * attributes. * * @param attributes the preset attributes * @return the created consumer * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request * due to some internal error */ StreamingConsumer createStreamingConsumer(KeyValue attributes); /** * Gets a lightweight {@code ResourceManager} instance from the specified {@code MessagingAccessPoint}. * * @return the resource manger * @throws OMSRuntimeException if the {@code MessagingAccessPoint} fails to handle this request * due to some internal error */ ResourceManager resourceManager(); }Copy the code- MessagingAccessPoint is similar to the OMS factory method, which aggregates methods for creating various objects, such as createProducer, createPushConsumer, createPullConsumer, and resourceManager
OMS
openmessaging-java/openmessaging-api/src/main/java/io/openmessaging/OMS.java
/**
* The oms class provides some static methods to create a {@code MessagingAccessPoint}
* from the specified OMS driver url and some useful util methods.
* <p>
* The complete OMS driver URL syntax is:
* <p>
* {@literal oms:<driver_type>://[account_id@]host1[:port1][,host2[:port2],...[,hostN[:portN]]]/<region>}
* <p>
* The first part of the URL specifies which OMS implementation is to be used, rocketmq is a
* optional driver type.
* <p>
* The brackets indicate that the extra access points are optional, and a correct OMS driver url
* needs at least one access point, whichconsists of hostname and port, Like localhost:8081. ** @version OMS 1.0.0 * @see ResourceManager * @since OMS 1.0.0 */ public final class OMS {/** * Returns a {@code MessagingAccessPoint} instance from the specified OMS driver url. * * @param url the specified OMS driver url * @return a {@code MessagingAccessPoint} instance
* @throws OMSRuntimeException ifthe factory fails to create a {@code MessagingAccessPoint} due to some driver url * some syntax error or internal error. */ public static MessagingAccessPoint getMessagingAccessPoint(String url) {return getMessagingAccessPoint(url, OMS.newKeyValue());
}
/**
* Returns a {@code MessagingAccessPoint} instance from the specified OMS driver url
* with some preset attributes, which will be passed to MessagingAccessPoint's implementation * class as a unique constructor parameter. * * There are some standard attributes defined by OMS for this method, * the same as {@link MessagingAccessPoint#attributes()} * * @param url the specified OMS driver url * @return a {@code MessagingAccessPoint} instance * @throws OMSRuntimeException if the factory fails to create a {@code MessagingAccessPoint} due to some driver url * some syntax error or internal error. */ public static MessagingAccessPoint getMessagingAccessPoint(String url, KeyValue attributes) { return MessagingAccessPointAdapter.getMessagingAccessPoint(url, attributes); } /** * Returns a default and internal {@code KeyValue} implementation instance. * * @return a {@code KeyValue} instance */ public static KeyValue newKeyValue() { return new DefaultKeyValue(); } /** * The version format is X.Y.Z (Major.Minor.Patch), a pre-release version may be denoted by appending a hyphen and a * series of dot-separated identifiers immediately following the patch version, like X.Y.Z-alpha. * * * OMS version follows semver scheme partially. * * @see http://semver.org */ public static String specVersion = "UnKnown"; static { InputStream stream = OMS.class.getClassLoader().getResourceAsStream("oms.spec.properties"); try { if (stream ! = null) { Properties properties = new Properties(); properties.load(stream); specVersion = String.valueOf(properties.get("version")); } } catch (IOException ignore) { } } private OMS() { } }
Copy the code- OMS can be understood as the total entry, a static factory method for MessagingAccessPoint
- It resolves information such as access addresses based on urls, such as oms: Rocketmq :// [email protected]/us-east:default_space
ResourceManager
openmessaging-java/openmessaging-api/src/main/java/io/openmessaging/ResourceManager.java
/**
* The {@code ResourceManager} is to provide a unified interface of resource management,
* allowing developers to manage the namespace, queue and routing resources.
* <p>
* Create, set. get and delete are the four basic operations of {@code ResourceManager}. * <p> * {@code ResourceManager} also supports dynamic fetch and update of resource attributes. * <p> * {@link MessagingAccessPoint#resourceManager()} ()} is the unique method to obtain a {@code ResourceManager}* instance. Changes made through this instance will immediately apply to the message-oriented middleware (MOM) behind * {@code MessagingAccessPoint}. ** @version OMS 1.0.0 * @since OMS 1.0.0 */ public interface ResourceManager {/** * Creates a {@code Namespace} resource with some preset attributes. * <p> * A namespace wraps the OMS resourcesinan abstract concept that makes it appear to the users * within the namespace that they have their own isolated instance of the global OMS resources. * * @param nsName the name of the new namespace * @param attributes the preset attributes */ void createNamespace(String nsName, KeyValue attributes); /** * Sets the attributes of the specific namespace, the old attributes will be replaced * by the provided attributes, only the provided key will be updated. * * @param nsName the specific namespace * @param attributes the new attributes * @throws OMSResourceNotExistExceptionif the specific namespace does not exist
*/
void setNamespaceAttributes(String nsName, KeyValue attributes) throws OMSResourceNotExistException;
/**
* Gets the attributes of the specific namespace.
*
* @param nsName the specific namespace
* @return the attributes of namespace
* @throws OMSResourceNotExistException if the specific namespace does not exist
*/
KeyValue getNamespaceAttributes(String nsName) throws OMSResourceNotExistException;
/**
* Deletes an existing namespace resource.
*
* @param nsName the namespace needs to be deleted
* @throws OMSResourceNotExistException if the specified namespace does not exist
*/
void deleteNamespace(String nsName) throws OMSResourceNotExistException;
/**
* Gets the namespace list in the current {@code MessagingAccessPoint}.
*
* @return the list of all namespaces
*/
List<String> listNamespaces();
/**
* Creates a {@code Queue} resource in the configured namespace with some preset attributes.
* <p>
* The standard OMS {@code Queue} schema must start with the {@code Namespace} prefix:
* <p>
* {@literal <namespace_name>://<queue_name>}
*
* @param queueName the name of the new queue
* @param attributes the preset attributes
* @throws OMSResourceNotExistException ifthe configured namespace does not exist or specified queue name is * not available */ void createQueue(String queueName, KeyValue attributes) throws OMSResourceNotExistException; /** * Sets the attributes of the specified queue, the old attributes will be replaced * by the provided attributes, only the provided key will be updated. * * @param queueName the queue name * @param attributes the new attributes * @throws OMSResourceNotExistExceptionif the specified queue or namespace does not exist
*/
void setQueueAttributes(String queueName, KeyValue attributes) throws OMSResourceNotExistException;
/**
* Gets the attributes of the specified queue.
*
* @param queueName the queue name
* @return the attributes of namespace
* @throws OMSResourceNotExistException if the specified queue or namespace does not exist
*/
KeyValue getQueueAttributes(String queueName) throws OMSResourceNotExistException;
/**
* Deletes an existing queue resource.
*
* @param queueName the queue needs to be deleted
* @throws OMSResourceNotExistException if the specified queue or namespace does not exist
*/
void deleteQueue(String queueName) throws OMSResourceNotExistException;
/**
* Gets the queue list in the specific namespace.
*
* @param nsName the specific namespace
* @return the list of all queues
* @throws OMSResourceNotExistException if the specific namespace does not exist
*/
List<String> listQueues(String nsName) throws OMSResourceNotExistException;
/**
* Creates a {@code Routing} resource in the configured namespace with some preset attributes.
* <p>
* The standard OMS {@code Routing} schema must start with the {@code Namespace} prefix:
* <p>
* {@literal <namespace_name>://<routing_name>}
*
* @param routingName the name of the new routing
* @param attributes the preset attributes
* @throws OMSResourceNotExistException if the configured namespace does not exist or specified routing name is not
* available
*/
void createRouting(String routingName, KeyValue attributes) throws OMSResourceNotExistException;
/**
* Sets the attributes of the specified routing, the old attributes will be replaced
* by the provided attributes, only the provided key will be updated.
*
* @param routingName the routing name
* @param attributes the new attributes
* @throws OMSResourceNotExistException if the specified routing or namespace does not exist
*/
void setRoutingAttributes(String routingName, KeyValue attributes) throws OMSResourceNotExistException;
/**
* Gets the attributes of the specified routing.
*
* @param routingName the routing name
* @return the attributes of routing
* @throws OMSResourceNotExistException if the specified routing or namespace does not exist
*/
KeyValue getRoutingAttributes(String routingName) throws OMSResourceNotExistException;
/**
* Deletes an existing routing resource.
*
* @param routingName the routing needs to be deleted
* @throws OMSResourceNotExistException if the specified routing or namespace does not exist
*/
void deleteRouting(String routingName) throws OMSResourceNotExistException;
/**
* Gets the routing list in the specific namespace.
*
* @param nsName the specific namespace
* @return the list of all routings
* @throws OMSResourceNotExistException if the specific namespace does not exist
*/
List<String> listRoutings(String nsName) throws OMSResourceNotExistException;
/**
* Gets the stream list behind the specified queue.
*
* @param queueName the queue name
* @return the list of all streams
* @throws OMSResourceNotExistException if the specified queue or namespace does not exist
*/
List<String> listStreams(String queueName) throws OMSResourceNotExistException;
/**
* Updates some system headers of a message in the configured namespace.
* <p>
* Below system headers are allowed to be changed dynamically:
* <ul>
* <li>{@link Message.BuiltinKeys#START_TIME}</li>
* <li>{@link Message.BuiltinKeys#STOP_TIME}</li>
* <li>{@link Message.BuiltinKeys#TIMEOUT}</li>
* <li>{@link Message.BuiltinKeys#PRIORITY}</li>
* <li>{@link Message.BuiltinKeys#SCHEDULE_EXPRESSION}</li>
* </ul>
*
* @param queueName the specific queue the message resides in
* @param messageId the id of message
* @param headers the new headers
* @throws OMSResourceNotExistException if the specified queue, namespace or message does not exist
*/
void updateMessage(String queueName, String messageId, KeyValue headers) throws OMSResourceNotExistException;
}
Copy the code- Provides operations such as namespace, queue, and routing
- Namespace operations include createNamespace, setNamespaceAttributes, deleteNamespace, listNamespaces, and getNamespaceAttributes
- Queue-related operations are createQueue, setQueueAttributes, getQueueAttributes, deleteQueue, listQueues
- Routing related operations are createRouting, setRoutingAttributes, getRoutingAttributes, deleteRouting, and listRoutings
- ListStreams and updateMessage methods are also provided
summary
Openmessaging MessagingAccessPoint aggregates the relevant access entry points, which can be understood as a facade. In addition, OMS is the static factory of MessagingAccessPoint. ResourceManager is similar to kafkaAdmin for managing namespaces, queues, and routing.
doc
- openmessaging.cloud
- OpenMessaging Runtime Interface for Java