diff --git a/README.md b/README.md index b9cbb91..587fb81 100644 --- a/README.md +++ b/README.md @@ -1,24 +1,55 @@ # Introduction -NATS is simple and secure messaging made for developers and operators who want to spend more time developing applications than worrying about a messaging system. +## The Importance of Messaging + +Developing and deploying applications that communicate in distributed systems +can be complex and difficult. A communication infrastructure should provide +features to make this easier, including multiple messaging patterns bundled +into one technology, location transparency, the decoupling of data producers +and consumers, and asynchronous communications to build event driven +applications. This can drive services/micro-services based workloads and take +the place of things like a service mesh, and also drive observability, all +from the same technology. + +### Distributed Computing Needs of Today + +A modern messaging system needs to support multiple communication patterns, be +secure by default, support multiple qualities of service, and provide secure +multi-tenancy for a truly shared infrastructure. A modern system needs to include: + +* Secure by default communications for microservices, edge platforms and devices +* Secure multi-tenancy in a single distributed communication technology +* Decoupled from IP for addressing and security +* Resiliency with an emphasis on the health of the system as a whole +* Ease of use for agile development, CI/CD, and operations, at scale +* Highly scalable with built-in load balancing and no config auto-scaling +* Consistent identity and security mechanisms from edge devices to backend services + +## NATS + +NATS was built to meet the distributed computing needs of today and tomorrow. +NATS is simple and secure messaging made for developers and operators who want +to spend more time developing modern applications and services than worrying +about a distributed communication system. * Easy to use for developers and operators * High-Performance * Always on and available * Extremely lightweight * At Most Once (NATS) or At Least Once Delivery (NATS Streaming) -* Common Messaging Pattern Support +* Common Messaging Pattern Support (Scalable Services, Event/Data Streams) * Client support for over 30 different programming languages * Cloud Native, a CNCF project with Kubernetes and Prometheus integrations -NATS is used for: +### Use Cases + +NATS can run anywhere, from large servers and cloud instances, through edge +gateways and even IoT devices. Use cases for NATS include: * Cloud Messaging -* Microservices + * Services (microservices, service mesh) + * Event/Data Streaming (observability, analytics, ML/AI) * Command and Control -* Service Discovery -* Event Streaming -* IoT and Edge -* Telemetry / Sensor Data -* Mobile and Big Data -* Augmenting or Replacing Legacy Messaging + * IoT and Edge + * Telemetry / Sensor Data / Command and Control +* Augmenting or Replacing Legacy Messaging Systems diff --git a/SUMMARY.md b/SUMMARY.md index eb642bc..4349007 100644 --- a/SUMMARY.md +++ b/SUMMARY.md @@ -3,18 +3,20 @@ * [Introduction](README.md) * [What's New in 2.0](whats_new/whats_new_20.md) +## Concepts + +* [What is NATS](developer/concepts/intro.md) +* [Subject-Based Messaging](developer/concepts/subjects.md) +* [Publish-Subscribe](developer/concepts/pubsub.md) +* [Request-Reply](developer/concepts/reqreply.md) +* [Queue Groups](developer/concepts/queue.md) +* [Acknowledgements](developer/concepts/acks.md) +* [Sequence Numbers](developer/concepts/seq_num.md) + ## Developing With NATS * [Introduction](developer/README.md) -* [Concepts](developer/concepts/intro.md) - * [Subject-Based Messaging](developer/concepts/subjects.md) - * [Publish-Subscribe](developer/concepts/pubsub.md) - * [Request-Reply](developer/concepts/reqreply.md) - * [Queue Groups](developer/concepts/queue.md) - * [Acknowledgements](developer/concepts/acks.md) - * [Sequence Numbers](developer/concepts/seq_num.md) - * [Connecting](developer/connecting/intro.md) * [Connecting to the Default Server](developer/connecting/default_server.md) * [Connecting to a Specific Server](developer/connecting/specific_server.md) @@ -120,6 +122,32 @@ * [Tutorial](nats_tools/nats_top/tutorial.md) * [nats-bench](nats_tools/natsbench.md) +## NATS Streaming Concepts + +* [Introduction](nats_streaming/intro.md) + * [Relation to NATS](nats_streaming/relation-to-nats.md) +* [Client Connections](nats_streaming/client-connections.md) +* [Channels](nats_streaming/channels/channels.md) + * [Message Log](nats_streaming/channels/message-log.md) + * [Subscriptions](nats_streaming/channels/subscriptions/subscriptions.md) + * [Regular](nats_streaming/channels/subscriptions/regular.md) + * [Durable](nats_streaming/channels/subscriptions/durable.md) + * [Queue Group](nats_streaming/channels/subscriptions/queue-group.md) + * [Redelivery](nats_streaming/channels/subscriptions/redelivery.md) +* [Store Interface](nats_streaming/store-interface.md) +* [Store Encryption](nats_streaming/store-encryption.md) +* [Clustering](nats_streaming/clustering/clustering.md) + * [Supported Stores](nats_streaming/clustering/supported-stores.md) + * [Configuration](nats_streaming/clustering/configuration.md) + * [Auto Configuration](nats_streaming/clustering/auto-configuration.md) + * [Containers](nats_streaming/clustering/containers.md) +* [Fault Tolerance](nats_streaming/fault-tolerance/ft.md) + * [Active Server](nats_streaming/fault-tolerance/active-server.md) + * [Standby Servers](nats_streaming/fault-tolerance/standby-server.md) + * [Shared State](nats_streaming/fault-tolerance/shared-state.md) + * [Failover](nats_streaming/fault-tolerance/failover.md) +* [Partitioning](nats_streaming/partitioning.md) + ## Developing With NATS Streaming * [Introduction](developer/streaming/README.md) @@ -129,14 +157,21 @@ * [Durable Subscriptions](developer/streaming/durables.md) * [Queue Subscriptions](developer/streaming/queues.md) * [Acknowledgements](developer/streaming/acks.md) +* [The Streaming Protocol](developer/streaming/protocol.md) ## NATS Streaming Server -* [Basics](nats_streaming/nats-streaming-intro.md) -* [Installing](nats_streaming/nats-streaming-install.md) - * [Securing](nats_streaming/nats-streaming-tls.md) -* [Protocol](nats_streaming/nats-streaming-protocol.md) -* [Docker Swarm](nats_streaming/nats-streaming-swarm.md) +* [Installing](nats_streaming/gettingstarted/install.md) +* [Running](nats_streaming/gettingstarted/run.md) +* [Configuring](nats_streaming/gettingstarted/configuring.md) +* [Securing](nats_streaming/gettingstarted/tls.md) +* [Process Signaling](nats_streaming/gettingstarted/process-signaling.md) +* [Windows Service](nats_streaming/gettingstarted/windows-service.md) +* [Embedding NATS Streaming Server](developer/streaming/embedding.md) +* [Docker Swarm](nats_streaming/swarm.md) +* [Monitoring](nats_streaming/monitoring/monitoring.md) + * [Enabling](nats_streaming/monitoring/enabling.md) + * [Endpoints](nats_streaming/monitoring/endpoints.md) ## NATS Protocol diff --git a/book.json b/book.json index b9ab25f..d7cdf3a 100644 --- a/book.json +++ b/book.json @@ -1,19 +1,20 @@ { - "title" : "NATS", - "description": "Administrative, developer and conceptual documentation for the NATS messaging system.", - "author": "The NATS Maintainers", - "plugins" : [ - "prism", - "-highlight", - "include-html", - "toggle-chapters" - ], - "pluginsConfig": { - "prism": { - "lang": { - "ascii": "markup", - "text": "markup" - } - } + "title" : "NATS", + "description": "Administrative, developer and conceptual documentation for the NATS messaging system.", + "author": "The NATS Maintainers", + "plugins" : [ + "prism", + "-highlight", + "include-html", + "toggle-chapters", + "anchors" + ], + "pluginsConfig": { + "prism": { + "lang": { + "ascii": "markup", + "text": "markup" } + } + } } \ No newline at end of file diff --git a/developer/concepts/intro.md b/developer/concepts/intro.md index cacba8e..963a8ff 100644 --- a/developer/concepts/intro.md +++ b/developer/concepts/intro.md @@ -1,4 +1,4 @@ -# NATS Messaging Concepts +# What is NATS NATS messaging involves the electronic exchange of data among computer applications and provides a layer between the application and the underlying physical network. Application data is encoded as a message and sent by a publisher. The message is received, decoded, and processed by one or more subscribers. diff --git a/developer/resources/ring_buffer.png b/developer/resources/ring_buffer.png deleted file mode 100644 index 44fd040..0000000 Binary files a/developer/resources/ring_buffer.png and /dev/null differ diff --git a/developer/resources/start_positions.png b/developer/resources/start_positions.png deleted file mode 100644 index 2dd7fa9..0000000 Binary files a/developer/resources/start_positions.png and /dev/null differ diff --git a/developer/streaming/README.md b/developer/streaming/README.md index afedfd2..cecafd2 100644 --- a/developer/streaming/README.md +++ b/developer/streaming/README.md @@ -68,15 +68,13 @@ distributed system. Where NATS provides at most once quality of service, streaming adds at least once. Streaming is implemented as a request-reply service on top of NATS. Streaming messages are encoded as protocol buffers, the streaming clients use NATS to talk to the streaming server. The streaming server organizes messages in channels and stores them in files and databases. ACKs are used to insure delivery in both directions. -> Sometimes the maintainers will refer to NATS as "nats core" and streaming as "stan" or "streaming." +> Sometimes the maintainers will refer to NATS as "nats core" and streaming as "stan" or "streaming". Messages to the streaming service are opaque byte arrays, just as they are with NATS. However, the streaming server protocol uses protocol buffers to wrap these byte arrays. So if you listen to the NATS traffic the messages will appear as protocol buffers, while the actual data sent and received will simply be byte arrays. -NATS streaming uses the concept of a channel to represent an ordered collection of messages. Clients send to and receive from channels instead of subjects. The subjects used by the streaming libraries and server are managed internally. Channels do not currently support wildcard. Channels aren’t raw subjects. Streaming isn’t raw NATS. The streaming libraries hide some of the differences. +NATS streaming uses the concept of a channel to represent an ordered collection of messages. Clients send to and receive from channels instead of subjects. The subjects used by the streaming libraries and server are managed internally. Channels do not currently support wildcards. Channels aren’t raw subjects. Streaming isn’t raw NATS. The streaming libraries hide some of the differences. -Think of channels as a ring buffer. Messages are added until the configured limit is reached. Old messages are removed to make room for new ones. Old messages can expire, based on configuration. Subscriptions don’t affect channel content. Subscriptions are like a cursor on the ring buffer. - -![ring buffer](../resources/ring_buffer.png) +Think of channels as a First In First Out (FIFO) queue. Messages are added until the configured limit is reached. Old messages can be set to expire based on configuration, making room for new messages. Subscriptions don’t affect channel content, that is, when a message is acknowledged, it is not removed from the channel. Positions in the channel are specified in multiple ways: @@ -84,9 +82,7 @@ Positions in the channel are specified in multiple ways: * Time * Time delta (converted to time on client) -New subscriptions can also specify last received to indicate they only want new messages. Sequence numbers are persistent, when message #1 goes away the oldest message is message #2. Trying to go to a position before the oldest message will be moved to the oldest message. - -![starting positions](../resources/start_positions.png) +New subscriptions can also specify last received to indicate they only want new messages. Sequence numbers are persistent so when message #1 goes away, the oldest message is then message #2. If you try to go to a position before the oldest message, you will be moved to the oldest message. ## Subscription Types diff --git a/developer/streaming/acks.md b/developer/streaming/acks.md index 72eeb94..c753258 100644 --- a/developer/streaming/acks.md +++ b/developer/streaming/acks.md @@ -14,7 +14,7 @@ sub, err := sc.Subscribe("foo", }, stan.SetManualAckMode(), stan.AckWait(aw)) ``` -## Max In Flight +# Max In Flight Subscribers can set max in flight to rate limit incoming messages. The server will send at most “max in flight” messages before receiving an acknowledgement. Setting max in flight to 1 insures every message is processed in order. diff --git a/developer/streaming/connecting.md b/developer/streaming/connecting.md index 20e83c2..eb12cb6 100644 --- a/developer/streaming/connecting.md +++ b/developer/streaming/connecting.md @@ -1,11 +1,22 @@ # Connecting to NATS Streaming -NATS Streaming is a service on top of NATS. To connect to the service you first connect to NATS and then use the client library to communicate with the server over your NATS connection. Most of the libraries provide a convenience mechanism for connecting in a single step. These convenience methods will take some NATS options, like the server, and perform the NATS connection first, then then run the protocol to connect to the streaming server. +First, it is recommended to understand the relation between Streaming and core NATS. You should familiarize yourself with the [concept](/nats_streaming/relation-to-nats.md). + +NATS Streaming is a service on top of NATS. To connect to the service you first connect to NATS and then use the client library to communicate with the server over your NATS connection. Most of the libraries provide a convenience mechanism for connecting in a single step. These convenience methods will take some NATS options, like the cluster ID, and perform the NATS connection first, then run the protocol to connect to the streaming server. Connecting to a streaming server requires a cluster id, defined by the server configuration, and a client ID defined by the client. +_Client ID should contain only alphanumeric characters, `-` or `_`_ + +Connecting to a server running locally on the default port is as simple as this: + ```go -sc, err := stan.Connect(clusterID, clientID, stan.NatsURL(“nats://localhost:4222”)) +sc, err := stan.Connect(clusterID, clientID) +``` + +If the server runs on port `1234`: +```go +sc, err := stan.Connect(clusterID, clientID, stan.NatsURL(“nats://localhost:1234)) ``` Sometimes you may want to provide NATS settings that aren't available in the streaming libraries connect method. Or, you may want to reuse a NATS connection instead of creating a new one. In this case the libraries generally provide a way to connect to streaming with an existing NATS connection: diff --git a/developer/streaming/durables.md b/developer/streaming/durables.md index e10023b..c5d2550 100644 --- a/developer/streaming/durables.md +++ b/developer/streaming/durables.md @@ -1,7 +1,13 @@ # Durable Subscriptions -Regular subscriptions remember their position while the client is connected. If the client disconnects the position is lost. Durable subscriptions remember their position even if the client is disconnected. Durable subscriptions identify themselves with a name. Connect and disconnect won’t affect the durable subscriptions position in the channel. Unsubscribe will clear the durable subscription. +Regular subscriptions remember their position while the client is connected. If the client disconnects the position is lost. Durable subscriptions remember their position even if the client is disconnected. + +Durable subscriptions identify themselves with a name. Connect and disconnect won’t affect the durable subscriptions position in the channel. ```go sc.Subscribe("foo", func(m *stan.Msg) {...}, stan.DurableName("my-durable")) ``` + +Unsubscribe will cause the server to completely remove the durable subscription. + +Check the [concepts](/nats_streaming/channels/subscriptions/durable.md) section for more information. \ No newline at end of file diff --git a/developer/streaming/embedding.md b/developer/streaming/embedding.md new file mode 100644 index 0000000..3ab441e --- /dev/null +++ b/developer/streaming/embedding.md @@ -0,0 +1,111 @@ +# Embedding NATS Streaming + +Embedding a NATS Streaming Server in your own code is easy. Simply import: + +``` + stand "github.com/nats-io/nats-streaming-server/server" +``` + +(Note: we chose `stand` here, but you don't have to use that name) + +Then if you want to use default options, it is as simple as doing: + +``` + s, err := stand.RunServer("mystreamingserver") +``` + +If you want a more advance configuration, then you need to pass options. For instance, let's start the server with a file store instead of memory. + +First import the stores package so we have access to the store type. + +``` + stores "github.com/nats-io/nats-streaming-server/stores" +``` + +Then get the default options and override some of them: + +``` + opts := stand.GetDefaultOptions() + opts.StoreType = stores.TypeFile + opts.FilestoreDir = "datastore" + s, err := stand.RunServerWithOpts(opts, nil) +``` + +However, since the NATS Streaming Server project vendors NATS Server that is uses as the communication layer with its clients and other servers in the cluster, there are some limitations. + +If you were to import `github.com/nats-io/nats-server/server`, instantiate a NATS `Options` structure, configure it and pass it to the second argument of `RunServerWithOpts`, you would get a compiler error. For instance doing this does not work: + +``` +import ( + natsd "github.com/nats-io/nats-server/server" + stand "github.com/nats-io/nats-streaming-server/server" + stores "github.com/nats-io/nats-streaming-server/stores" +) + +(...) + + nopts := &natsd.Options{} + nopts.Port = 4223 + + s, err := stand.RunServerWithOpts(nil, nopts) +``` + +You would get: + +``` +./myapp.go:36:35: cannot use nopts (type *"myapp/vendor/github.com/nats-io/nats-server/server".Options) as type *"myapp/vendor/github.com/nats-io/nats-streaming-server/vendor/github.com/nats-io/gnatsd/server".Options in argument to "myapp/vendor/github.com/nats-io/nats-streaming-server/server".RunServerWithOpts +``` + +To workaround this issue, the NATS Streaming Server package provides a function `NewNATSOptions()` that is suitable for this approach: + +``` + nopts := stand.NewNATSOptions() + nopts.Port = 4223 + + s, err := stand.RunServerWithOpts(nil, nopts) +``` + +That will work. + +But, if you want to do advanced NATS configuration that requires types or interfaces that belong to the NATS Server package, then this approach won't work. In this case, you need to run the NATS Server independently and have the NATS Streaming Server connect to it. + +``` + // This configure the NATS Server using natsd package + nopts := &natsd.Options{} + nopts.HTTPPort = 8222 + nopts.Port = 4223 + + // Setting a customer client authentication requires the NATS Server Authentication interface. + nopts.CustomClientAuthentication = &myCustomClientAuth{} + + // Create the NATS Server + ns := natsd.New(nopts) + + // Start it as a go routine + go ns.Start() + + // Wait for it to be able to accept connections + if !ns.ReadyForConnections(10 * time.Second) { + panic("not able to start") + } + + // Get NATS Streaming Server default options + opts := stand.GetDefaultOptions() + + // Point to the NATS Server with host/port used above + opts.NATSServerURL = "nats://localhost:4223" + + // Now we want to setup the monitoring port for NATS Streaming. + // We still need NATS Options to do so, so create NATS Options + // using the NewNATSOptions() from the streaming server package. + snopts := stand.NewNATSOptions() + snopts.HTTPPort = 8223 + + // Now run the server with the streaming and streaming/nats options. + s, err := stand.RunServerWithOpts(opts, snopts) + if err != nil { + panic(err) + } +``` + +The above process may seem involved, but only if you use very advanced NATS Server options. \ No newline at end of file diff --git a/developer/streaming/protocol.md b/developer/streaming/protocol.md new file mode 100644 index 0000000..b9901e4 --- /dev/null +++ b/developer/streaming/protocol.md @@ -0,0 +1,265 @@ +# Writing your own client library + +You can find a list of all supported client libraries [here](https://nats.io/download/). There are also links to community contributed clients. + +In the event you would want to write your own NATS Streaming library, you could have a look at existing libraries to understand the flow. But you need to use [Google Protocol Buffers](https://developers.google.com/protocol-buffers/) to exchange protocols between the client and the server. + +## NATS Streaming Protocol + +The NATS streaming protocol sits atop the core NATS protocol and uses [Google's Protocol Buffers](https://developers.google.com/protocol-buffers/). Protocol buffer messages are marshaled into bytes and published as NATS messages on specific subjects described below. In communicating with the NATS Streaming Server, the NATS request/reply pattern is used for all protocol messages that have a corresponding reply. + +### NATS streaming protocol conventions + +**Subject names**: Subject names, including reply subject (INBOX) names, are case-sensitive and must be non-empty alphanumeric strings with no embedded whitespace, and optionally token-delimited using the dot character (`.`), e.g.: + +`FOO`, `BAR`, `foo.bar`, `foo.BAR`, `FOO.BAR` and `FOO.BAR.BAZ` are all valid subject names + +`FOO. BAR`, `foo. .bar` and`foo..bar` are *not- valid subject names + +**Wildcards**: NATS streaming does **not** support wildcards in subject subscriptions + +**Protocol definition**: The fields of NATS streaming protocol messages are defined in the NATS streaming client [protocol file](https://github.com/nats-io/stan.go/blob/master/pb/protocol.proto). + +### NATS streaming protocol messages + +The following table briefly describes the NATS streaming protocol messages. + +Click the name to see more detailed information, including usage: + +#### Protocols + +| Message Name | Sent By | Description +| ------------------------------------------------- |:--------|:-------------------------------------------- +| [`ConnectRequest`](#connectrequest) | Client | Request to connect to the NATS Streaming Server +| [`ConnectResponse`](#connectresponse) | Server | Result of a connection request +| [`SubscriptionRequest`](#subscriptionrequest) | Client | Request sent to subscribe and retrieve data +| [`SubscriptionResponse`](#subscriptionresponse) | Server | Result of a subscription request +| [`UnsubscribeRequest`](#unsubscriberequest) | Client | Unsubscribe from a subject +| [`PubMsg`](#pubmsg) | Client | Publish a message to a subject +| [`PubAck`](#puback) | Server | An acknowledgement that a published message has been processed on the server +| [`MsgProto`](#msgproto) | Server | A message from the NATS Streaming Server to a subscribing client +| [`Ack`](#ack) | Client | Acknowledges that a message has been received +| [`Ping`](#ping) | Client | Ping sent to server to detect connection loss +| [`PingResponse`](#pingresponse) | Server | Result of a Ping +| [`CloseRequest`](#closerequest) | Client | Request sent to close the connection to the NATS Streaming Server +| [`CloseResp`](#closeresponse) | Server | Result of the close request + +The following sections explain each protocol message. + +#### ConnectRequest + +##### Description + +A connection request is sent when a streaming client connects to the NATS Streaming Server. The connection request contains a unique identifier representing the client, and an inbox subject the client will listen on for incoming heartbeats. The identifier **must** be unique; a connection attempt with an identifier currently in use will fail. The inbox subject is the subject where the client receives incoming heartbeats, and responds by publishing an empty NATS message to the reply subject, indicating it is alive. The NATS Streaming Server will return a [ConnectResponse](#connectresponse) message to the reply subject specified in the NATS request message. + +More advanced libraries can set the protocol to 1 and send a connection ID which in combination with ping interval and ping max out allows the library to detect that the connection to the server is lost. + +This request is published to a subject comprised of the `.cluster-id`, for example, if a NATS Streaming Server was started with a cluster-id of `mycluster`, and the default prefix was used, the client publishes to `_STAN.discover.mycluster` + +##### Message Structure + +- `clientID`: A unique identifier for a client +- `heartbeatInbox`: An inbox to which the NATS Streaming Server will send heartbeats for the client to process +- `protocol`: Protocol the client is at +- `connID`: Connection ID, a way to uniquely identify a connection (no connection should ever have the same) +- `pingInterval`: Interval at which client wishes to send PINGs (expressed in seconds) +- `pingMaxOut`: Maximum number of PINGs without a response after which the connection can be considered lost + +[Back to table](#protocols) + + +#### ConnectResponse + +##### Description + +After a `ConnectRequest` is published, the NATS Streaming Server responds with this message on the reply subject of the underlying NATS request. The NATS Streaming Server requires the client to make requests and publish messages on certain subjects (described above), and when a connection is successful, the client saves the information returned to be used in sending other NATS streaming protocol messages. In the event the connection was not successful, an error is returned in the `error` field. + +##### Message Structure + +- `pubPrefix`: Prefix to use when publishing +- `subRequests`: Subject used for subscription requests +- `unsubRequests`: Subject used for unsubscribe requests +- `closeRequests`: Subject for closing a connection +- `error`: An error string, which will be empty/omitted upon success +- `subCloseRequests`: Subject to use for subscription close requests +- `pingRequests`: Subject to use for PING requests +- `pingInterval`: Interval at which client should send PINGs (expressed in seconds). +- `pingMaxOut`: Maximum number of PINGs without a response after which the connection can be considered lost +- `protocol`: Protocol version the server is at +- `publicKey`: Reserved for future use + +[Back to table](#protocols) + + +#### SubscriptionRequest + +##### Description + +A `SubscriptionRequest` is published on the subject returned in the `subRequests` field of a [ConnectResponse](#connectresponse), and creates a subscription to a subject on the NATS Streaming Server. This will return a [SubscriptionResponse](#subscriptionresponse) message to the reply subject specified in the NATS protocol request message. + +##### Message Structure + +- `clientID`: Client ID originally provided in the [ConnectRequest](#connectrequest) +- `subject`: Formal subject to subscribe to, e.g. foo.bar +- `qGroup`: Optional queue group +- `inbox`: Inbox subject to deliver messages on +- `maxInFlight`: Maximum inflight messages without an acknowledgement allowed +- `ackWaitInSecs`: Timeout for receiving an acknowledgement from the client +- `durableName`: Optional durable name which survives client restarts +- `startPosition`: An enumerated type specifying the point in history to start replaying data +- `startSequence`: Optional start sequence number +- `startTimeDelta`: Optional start time + +##### StartPosition enumeration + +- `NewOnly`: Send only new messages +- `LastReceived`: Send only the last received message +- `TimeDeltaStart`: Send messages from duration specified in the `startTimeDelta` field. +- `SequenceStart`: Send messages starting from the sequence in the `startSequence` field. +- `First`: Send all available messages + +[Back to table](#protocols) + + +#### SubscriptionResponse + +##### Description + +The `SubscriptionResponse` message is the response from the `SubscriptionRequest`. After a client has processed an incoming [MsgProto](#msgproto) message, it must send an acknowledgement to the `ackInbox` subject provided here. + +##### Message Structure + +- `ackInbox`: subject the client sends message acknowledgements to the NATS Streaming Server +- `error`: error string, empty/omitted if no error + +[Back to table](#protocols) + + +#### UnsubscribeRequest + +##### Description + +The `UnsubscribeRequest` closes or unsubcribes the subscription from the specified subject. The inbox specified is the `inbox` returned from the NATS Streaming Server in the `SubscriptionResponse`. Depending on which subject this request is sent, the action will result in close (if sent to subject `subCloseRequests`) or unsubscribe (if sent to subject `unsubRequests`) + +##### Message Structure + +- `clientID`: Client ID originally provided in the [ConnectRequest](#connectrequest) +- `subject`: Subject for the subscription +- `inbox`: Inbox subject to identify subscription +- `durableName`: Optional durable name which survives client restarts + +[Back to table](#protocols) + + +#### PubMsg + +##### Description + +The `PubMsg` protocol message is published from a client to the NATS Streaming Server. The GUID must be unique, and is returned in the [PubAck](#puback) message to correlate the success or failure of storing this particular message. + +##### Message Structure + +- `clientID`: Client ID originally provided in the [ConnectRequest](#connectrequest) +- `guid`: a guid generated for this particular message +- `subject`: subject +- `data`: payload +- `connID`: Connection ID. For servers that know about this field, clientID can be omitted + +[Back to table](#protocols) + + +#### PubAck + +##### Description + +The `PubAck` message is an acknowledgement from the NATS Streaming Server that a message has been processed. The message arrives on the subject specified on the reply subject of the NATS message the `PubMsg` was published on. The GUID is the same GUID used in the `PubMsg` being acknowledged. If an error string is present, the message was not persisted by the NATS Streaming Server and no guarantees regarding persistence are honored. `PubAck` messages may be handled asynchronously from their corresponding `PubMsg` in the client. + +##### Message Structure + +- `guid`: GUID of the message being acknowledged by the NATS Streaming Server +- `error`: An error string, empty/omitted if no error + +[Back to table](#protocols) + + +#### MsgProto + +##### Description + +The `MsgProto` message is received by client from the NATS Streaming Server, containing the payload of messages sent by a publisher. A `MsgProto` message that is not acknowledged with an [Ack](#ack) message within the duration specified by the `ackWaitInSecs` field of the subscription request will be redelivered. + +##### Message Structure + +- `sequence`: Globally ordered sequence number for the subject's channel +- `subject`: Subject +- `data`: Payload +- `timestamp`: Time the message was stored in the server. +- `redelivered`: Flag specifying if the message is being redelivered + +[Back to table](#protocols) + + +#### Ack + +##### Description + +An `Ack` message is an acknowledgement from the client that a [MsgProto](#msgproto) message has been considered received. It is published to the `ackInbox` field of the [SubscriptionResponse](#subscriptionresponse). + +##### Message Structure + +- `subject`: Subject of the message being acknowledged +- `sequence`: Sequence of the message being acknowledged + +[Back to table](#protocols) + + +#### Ping + +##### Description + +A `Ping` message is sent to the server at configured interval to check that the connection ID is still valid. This should be used only if client is at protocol 1, and has sent a `connID` in the [ConnectRequest](#connectrequest) protocol. + +##### Message Structure + +- `connID`: The connection ID + +[Back to table](#protocols) + + +#### PingResponse + +##### Description + +This is a response from the server to a `Ping` from the client. If the content is not empty, it will be the error indicating to the client why the connection is no longer valid. + +##### Message Structure + +- `error`: Error string, empty/omitted if no error + +[Back to table](#protocols) + + +#### CloseRequest + +##### Description + +A `CloseRequest` message is published on the `closeRequests` subject from the [ConnectResponse](#connectresponse), and notifies the NATS Streaming Server that the client connection is closing, allowing the server to free up resources. This message should **always** be sent when a client is finished using a connection. + +##### Message Structure + +- `clientID`: Client ID originally provided in the [ConnectRequest](#connectrequest) + +[Back to table](#protocols) + + +#### CloseResponse + +##### Description + +The `CloseResponse` is sent by the NATS Streaming Server on the reply subject of the `CloseRequest` NATS message. This response contains any error that may have occurred with the corresponding close call. + +##### Message Structure + +- `error`: error string, empty/omitted if no error + +[Back to table](#protocols) \ No newline at end of file diff --git a/developer/streaming/publishing.md b/developer/streaming/publishing.md index b9eaa28..bc140fa 100644 --- a/developer/streaming/publishing.md +++ b/developer/streaming/publishing.md @@ -1,6 +1,6 @@ # Publishing to a Channel -The streaming client library can provide a method for publishing synchronously. .These publish methods block until the ACK is returned by the server. An error or exception is used to indicate a timeout or other error. +The streaming client library can provide a method for publishing synchronously. These publish methods block until the ACK is returned by the server. An error or exception is used to indicate a timeout or other error. ```go err := sc.Publish("foo", []byte("Hello World")) @@ -12,4 +12,10 @@ Streaming libraries can also provide a way to publish asynchronously. An ACK cal ackHandler := func(ackedNuid string, err error){ ... } nuid, err := sc.PublishAsync("foo", []byte("Hello World"), ackHandler) -``` \ No newline at end of file +``` + +Even in this mode, the call will still block if the library has a number of published messages without having received an ACK from the server. The default can be changed when creating the connection. + +```go +sc, err := sc.Connect(clusterID, clientName, stan.MaxPubAcksInflight(1000)) +``` diff --git a/developer/streaming/queues.md b/developer/streaming/queues.md index 0cad323..bb86eb4 100644 --- a/developer/streaming/queues.md +++ b/developer/streaming/queues.md @@ -1,11 +1,37 @@ # Queue Subscriptions -Queue subscriptions are created like other subscriptions with the addition of a queue name. All subscriptions, across clients, share the queue based on this unique name. Other subscriptions can receive messages independently of the queue groups. Unsubscribe removes a client from a group, the last unsubscribe kills the group. Max in flight is per subscription. +Queue subscriptions are created like other subscriptions with the addition of a queue name. ```go -qsub1, _ := sc.QueueSubscribe(channelID, +qsub1, _ := sc.QueueSubscribe(channelName, queueName, func(m *stan.Msg) {...}) -qsub2, _ := sc.QueueSubscribe(channelID, +qsub2, _ := sc.QueueSubscribe(channelName, queueName, func(m *stan.Msg) {...}) -``` \ No newline at end of file +``` + +Multiple subscriptions using the same channel and queue name are members of the same queue group. That means that if a message is published on that channel, only one member of the group receives the message. Other subscriptions receive messages independently of the queue groups, that is, a message is delivered to all subscriptions and one member of each queue group. + +To create a durable queue subscription, simply add a durable name: +```go +qsub, err := sc.QueueSubscribe(channelName, + queueName, func(m *stan.Msg) {...}, + stan.DurableName("durable-name")) +``` + +Subscriptions options apply to each member independently, notably, the `AckWait` and `MaxInflight`. Those two members of the same queue group use different options for redelivery and max inflight. +```go +qsub1, _ := sc.QueueSubscribe(channelName, + queueName, func(m *stan.Msg) {...}, + stan.AckWait(5*time.Second), + stan.MaxInflight(5)) + +qsub2, _ := sc.QueueSubscribe(channelName, + queueName, func(m *stan.Msg) {...}, + stan.AckWait(20*time.Second), + stan.MaxInflight(10)) +``` + +If the queue subscription is durable, only the last member calling `Unsubscribe()` will cause the durable queue group to be removed from the server. + +Check the [concepts](/nats_streaming/channels/subscriptions/queue-group.md) section for more information. \ No newline at end of file diff --git a/developer/streaming/receiving.md b/developer/streaming/receiving.md index 0811e06..f09d2c5 100644 --- a/developer/streaming/receiving.md +++ b/developer/streaming/receiving.md @@ -9,6 +9,10 @@ Subscriptions come in several forms: * Queue * Queue/Durable +For more details on the various types, check the [concepts](/nats_streaming/channels/subscriptions/subscriptions.md) section. + +***Note: message callbacks are invoked serially, one message at a time. If your application does not care about processing ordering and would prefer the messages to be dispatched concurrently, it is the application responsibility to move them to some internal queue to be picked up by threads/go routines.*** + Subscriptions set their starting position on creation using position or time. For example, in Go you can start at: * The last message received @@ -43,4 +47,26 @@ var startTime time.Time sub, err := sc.Subscribe("foo", func(m *stan.Msg) {...}, stan.StartAtTime(startTime)) -``` \ No newline at end of file +``` + +To set the delay after which the server should attempt to redeliver a message for which it has not receive an acknowledgment: + +```go +sub, err := sc.Subscribe("foo", + func(m *stan.Msg) {...}, + stan.AckWait(20*time.Second)) +``` + +When an application wishes to stop receiving, but want to maintain the connection opened, the subscription should be closed. There are two ways to stop a subscription, either "close" it, or "unsubscribe" it. For non durable subscriptions, this is equivalent since the subscription will be completely removed. For durable subscriptions, close means that the server will stop delivering, but remember the durable subscription. Unsubscribe, however, means that the server will remove the state of this subscription. + +To simply close: +```go +err := sub.Close() +``` + +To unsubscribe: +```go +err := sub.Unsubscribe() +``` + +_Note: If a connection is closed without explicitly closing the subscriptions, the subscriptions are implicitly closed, not unsubscribed._ diff --git a/docs/developer/concepts/acks.html b/docs/developer/concepts/acks.html index 45c1c70..fea2269 100644 --- a/docs/developer/concepts/acks.html +++ b/docs/developer/concepts/acks.html @@ -25,6 +25,10 @@ + + + + @@ -126,11 +130,743 @@ +
  • Concepts
    • + + + +
  • + + + + + What is NATS + + + + + +
    • + +
  • + + + + + Subject-Based Messaging + + + + + +
    • + +
  • + + + + + Publish-Subscribe + + + + + +
    • + +
  • + + + + + Request-Reply + + + + + +
    • + +
  • + + + + + Queue Groups + + + + + +
    • + +
  • + + + + + Acknowledgements + + + + + +
    • + +
  • + + + + + Sequence Numbers + + + + + +
    • + + + + +
  • Developing With NATS
    • + + + +
  • + + + + + Introduction + + + + + +
    • + +
  • + + + + + Connecting + + + + + + + +
    • + +
  • + + + + + Automatic Reconnections + + + + + + + +
    • + +
  • + + + + + Securing Connections + + + + + + + +
    • + +
  • + + + + + Receiving Messages + + + + + + + +
    • + +
  • + + + + + Sending Messages + + + + + + + +
    • + +
  • + + + + + Monitoring the Connection + + + + + + + +
    • + +
  • + + + + + Tutorials + + + + + + + +
    • + + + +
  • NATS Server
    • -
  • +
  • @@ -143,7 +879,7 @@
    • -
  • +
  • @@ -157,7 +893,7 @@