Updated doc

Added docker file
Fixed tests, all now run

Author: Stephen Asbury

Dockerfile

@@ -0,0 +1,20 @@
+FROM golang:1.12.4 AS builder
+
+WORKDIR /src/nats-replicator
+
+LABEL maintainer "Stephen Asbury <[email protected]>"
+
+COPY . .
+
+RUN go mod download
+RUN CGO_ENABLED=0 go build -v -a -tags netgo -installsuffix netgo -o /nats-replicator
+
+FROM alpine:3.9
+
+RUN mkdir -p /nats/bin && mkdir /nats/conf
+
+COPY --from=builder /nats-replicator /nats/bin/nats-replicator
+
+RUN ln -ns /nats/bin/nats-replicator /bin/nats-replicator
+
+ENTRYPOINT ["/bin/nats-replicator"]

README.md

@@ -11,11 +11,8 @@ This project implements a multi-connector bridge between NATS and NATS streaming
 
 ## WARNING DO NOT USE THIS SERVER YET
 
-* This is the initial checkin
-* This code was just started
-* Tests do not pass
-* Doc is not right
-* Docker is not provided
+* Stan incoming performance needs to be tweaked to allow multiple outstanding acks
+* Doc is a first draft
 
 ## Features
 

docs/buildandrun.md

@@ -28,8 +28,6 @@ Use `make test` to run the tests, and `make install` to install. The nats and na
 
 ## Docker
 
-!!! FIX THIS !!!
-
 You can build the docker image using:
 
 ```bash

docs/config.md

@@ -1,6 +1,6 @@
 # NATS-Replicator Configuration
 
-The bridge uses a single configuration file passed on the command line or environment variable. Configuration is organized into a root section and several blocks.
+The replicator uses a single configuration file passed on the command line or environment variable. Configuration is organized into a root section and several blocks.
 
 * [Specifying the Configuration File](#specify)
 * [Shared](#root)
@@ -24,10 +24,10 @@ include "./includes/connectors.conf"
 To set the configuration on the command line, use:
 
 ```bash
-% nats-kafka -c <path to config file>
+% nats-replicator -c <path to config file>
 ```
 
-To set the configuration file using an environment variable, export `NATS_KAFKA_BRIDGE_CONFIG` with the path to the configuration.
+To set the configuration file using an environment variable, export `NATS_REPLICATOR_CONFIG` with the path to the configuration.
 
 <a name="root"></a>
 
@@ -37,17 +37,15 @@ The root section:
 
 ```yaml
 reconnectinterval: 5000,
-connecttimeout: 5000,
 ```
 
 can currently contain settings for:
 
-* `reconnectinterval` - this value, in milliseconds, is the time used in between reconnection attempts for a connector when it fails. For example, if a connector loses access to NATS, the bridge will try to restart it every `reconnectinterval` milliseconds.
-* `connecttimeout` - this value, in milliseconds, is the time used when trying to connect to Kafka.
+* `reconnectinterval` - this value, in milliseconds, is the time used in between reconnection attempts for a connector when it fails. For example, if a connector loses access to NATS, the replicator will try to restart it every `reconnectinterval` milliseconds.
 
 ## TLS <a name="tls"></a>
 
-NATS, streaming, Kafka and HTTP configurations all take an optional TLS setting. The TLS configuration takes three possible settings:
+NATS, streaming and HTTP configurations all take an optional TLS setting. The TLS configuration takes three possible settings:
 
 * `root` - file path to a CA root certificate store, used for NATS connections
 * `cert` - file path to a server certificate, used for HTTPS monitoring and optionally for client side certificates with NATS
@@ -95,57 +93,69 @@ monitoring: {
 
 Is used to configure an HTTP or HTTPS port, as well as TLS settings when HTTPS is used.
 
-* `httphost` - the network interface to publish monitoring on, valid for HTTP or HTTPS. An empty value will tell the server to use all available network interfaces.
-* `httpport` - the port for HTTP monitoring, no TLS configuration is expected, a value of -1 will tell the server to use an ephemeral port, the port will be logged on startup.
+* `httphost` - the network interface to publish monitoring on, valid for HTTP or HTTPS. An empty value will tell the replicator to use all available network interfaces.
+* `httpport` - the port for HTTP monitoring, no TLS configuration is expected, a value of -1 will tell the replicator to use an ephemeral port, the port will be logged on startup.
 
 `2019/03/20 12:06:38.027822 [INF] starting http monitor on :59744`
 
 * `httpsport` - the port for HTTPS monitoring, a TLS configuration is expected, a value of -1 will tell the server to use an ephemeral port, the port will be logged on startup.
 * `tls` - a [TLS configuration](#tls).
 
-The `httpport` and `httpsport` settings are mutually exclusive, if both are set to a non-zero value the bridge will not start.
+The `httpport` and `httpsport` settings are mutually exclusive, if both are set to a non-zero value the replicator will not start.
 
 <a name="nats"></a>
 
 ## NATS
 
-The bridge makes a single connection to NATS. This connection is shared by all connectors. Configuration is through the `nats` section of the config file:
+The replicator can make multiple connections to NATS. Each connection has a name so that connectors can refer to it. Streaming connections also use the name of a NATS connection to refer to it. Configuration is through the `nats` section of the config file, which defines an array of connection parameters:
 
 ```yaml
-nats: {
-  Servers: ["localhost:4222"],
-  ConnectTimeout: 5000,
-  MaxReconnects: 5,
-  ReconnectWait: 5000,
-}
+nats: [
+  {
+    Name: "connection_one",
+    Servers: ["localhost:4222"],
+    ConnectTimeout: 5000,
+    MaxReconnects: 5,
+    ReconnectWait: 5000,
+  }
+]
 ```
 
 NATS can be configured with the following properties:
 
+* `name` - the unique name used to refer to this configuration/connection
 * `servers` - an array of server URLS
 * `connecttimeout` - the time, in milliseconds, to wait before failing to connect to the NATS server
 * `reconnectwait` - the time, in milliseconds, to wait between reconnect attempts
-* `maxreconnects` - the maximum number of reconnects to try before exiting the bridge with an error.
+* `maxreconnects` - the maximum number of reconnects to try before exiting the replicator with an error.
 * `tls` - (optional) [TLS configuration](#tls). If the NATS server uses unverified TLS with a valid certificate, this setting isn't required.
 * `usercredentials` - (optional) the path to a credentials file for connecting to NATs.
 
 <a name="stan"></a>
 
 ## NATS Streaming
 
-The bridge makes a single connection to a NATS streaming server. This connection is shared by all connectors. Configuration is through the `stan` section of the config file:
+The replicator can make multiple connections to streaming servers. Each streaming connection must refer to a NATS connection by name. This reduces the configuration clutter by keeping NATS parameters in the NATS section. Configuration is through the `stan` section of the config file, which defines an array:
 
 ```yaml
-stan: {
-  ClusterID: "test-cluster"
-  ClientID: "kafkabridge"
-}
+stan: [
+  {
+    Name: "stan_one",
+    NATSConnection: "connection_one",
+    ClusterID: "test-cluster"
+    ClientID: "replicator_one"
+  }
+]
 ```
 
+Multiple streaming connections could use the same NATS connection.
+
 NATS streaming can be configured with the following properties:
 
+* `name` - the unique name used to refer to this configuration/connection
+* `natsconnection` - the unique name of the nats connection to use for this streaming connection
 * `clusterid` - the cluster id for the NATS streaming server.
-* `clientid` - the client id for the bridge, shared by all connections.
+* `clientid` - the client id for the connection.
 * `pubackwait` - the time, in milliseconds, to wait before a publish fails due to a timeout.
 * `discoverprefix` - the discover prefix for the streaming server.
 * `maxpubacksinflight` - maximum pub ACK messages that can be in flight for this connection.
@@ -155,72 +165,51 @@ NATS streaming can be configured with the following properties:
 
 ## Connectors
 
-The final piece of the bridge configuration is the `connect` section. Connect specifies an array of connector configurations. All connector configs use the same format, relying on optional settings to determine what the do.
+The final piece of the replicator configuration is the `connect` section. Connect specifies an array of connector configurations. All connector configs use the same format, relying on optional settings to determine what the do.
 
 ```yaml
 connect: [
   {
-      type: "KafkaToNATS",
-      brokers: ["localhost:9092"]
-      id: "zip",
-      topic: "test",
-      subject: "one",
-  },{
-      type: "NATSToKafka",
-      brokers: ["localhost:9092"]
-      id: "zap",
-      topic: "test2",
-      subject: "two",
+      type: NATSToNATS,
+      id: "alpha",
+      IncomingSubject: "in",
+      IncomingConnection: "connection_one",
+      OutgoingSubject: "out",
+      OutgoingConnection: "connection_one",
   },
 ],
 ```
 
 The most important property in the connector configuration is the `type`. The type determines which kind of connector is instantiated. Available, uni-directional, types include:
 
-* `KafkaToNATS` - a topic to NATS connector
-* `KafkaToStan` - a topic to streaming connector
-* `NATSToKafka` - a streaming to topic connector
-* `STANToKafka` - a NATS to topic connector
+* `NATSToNATS` - a subject to subject connector
+* `NATSToStan` - a subject to streaming connector
+* `StanToNATS` - a streaming to subject connector
+* `STANToSTAN` - a streaming to streaming connector
 
 All connectors can have an optional id, which is used in monitoring:
 
 * `id` - (optional) user defined id that will tag the connection in monitoring JSON.
 
-For NATS connections, specify:
-
-* `subject` - the subject to subscribe/publish to, depending on the connections direction.
-* `natsqueue` - the queue group to use in subscriptions, this is optional but useful for load balancing.
+All connectors require a configuration for the connection to use:
 
-Keep in mind that NATS queue groups do not guarantee ordering, since the queue subscribers can be on different nats-servers in a cluster. So if you have to bridges running with connectors on the same NATS queue/subject pair and have a high message rate you may get messages in the Kafka topic out of order.
+* `incomingconnection` - the name of a NATS or streaming connection to subscribe to
+* `outgoingconnection` - the name of a NATS or streaming connection to publish to
 
-For streaming connections, there is a single required setting and several optional ones:
-
-* `channel` - the streaming channel to subscribe/publish to.
-* `durablename` - (optional) durable name for the streaming subscription (if appropriate.)
-* `startatsequence` - (optional) start position, use -1 for start with last received, 0 for deliver all available (the default.)
-* `startattime` - (optional) the start position as a time, in Unix seconds since the epoch, mutually exclusive with `startatsequence`.
-
-All connectors must specify Kafka connection properties, with a few optional settings available as well:
+For NATS connections, specify:
 
-* `brokers` - a string array of broker host:port settings
-* `topic` - the Kafka topic to listen/send to
-* `tls` - A tls config for the connection
-* `balancer` - required for a writer, should be "hash" or "leastbytes"
-* `groupid` - (exclusive with partition) used by the reader to set a group id
-* `partition` - (exclusive with groupid) used by the reader to set a partition
-* `minbytes` - (optional) used by Kafka readers to set the minimum bytes for a read
-* `maxbytes` - (optional) used by a Kafka reader to set the maximum bytes for a read
-* `keytype` - (optional) defines the way keys are assigned to messages coming from NATS (see below)
-* `keyvalue` - (optional) extra data that may be used depending on the key type
+* `incomingsubject` - the subject to subscribe to, depending on the connections direction.
+* `incomingqueuename` - the queue group to use in subscriptions, this is optional but useful for load balancing.
+* `outgoingsubject` - the subject to publish to, depending on the connections direction.
 
-Available key types are:
+Keep in mind that NATS queue groups do not guarantee ordering, since the queue subscribers can be on different nats-servers in a cluster. So if you have to replicators running with connectors on the same NATS queue/subject pair and have a high message rate you may get messages to the receiver "out of order." Also, note that there is no outgoing queue.
 
-* `fixed` - the value in the `keyvalue` field is assigned to all messages
-* `subject` - the subject the incoming NATS message was sent on is used as the key
-* `reply` - the reply-to sent with the incoming NATS messages is used as the key
-* `subjectre` - the value in the `keyvalue` field used as a regular expression and the first capture group, when matched to the subject, is used as the key
-* `replyre` - the value in the `keyvalue` field used as a regular expression and the first capture group, when matched to the reply-to, is used as the key
+These settings are directional depending so a `NATSToStan` connector would use an `incomingsubject` while a `StanToNATS` connector would use an `outgoingsubject`. Connectors ignore settings they don't need.
 
-If unset, an empty key is assigned during translation from NATS to Kafka. If the regex types are used and they don't match, an empty key is used.
+For streaming connections, the channel setting is required (directionality dependent), the others are optional:
 
-For nats streaming connections channel is treated as the subject and durable name is treated as the reply to, so that reply key type will use the durable name as the key.
+* `incomingchannel` - the streaming channel to subscribe to.
+* `outgoingchannel` - the streaming channel to publish to.
+* `incomingdurablename` - (optional) durable name for the streaming subscription (if appropriate.)
+* `incomingstartatsequence` - (optional) start position, use -1 for start with last received, 0 for deliver all available (the default.)
+* `incomingstartattime` - (optional) the start position as a time, in Unix seconds since the epoch, mutually exclusive with `startatsequence`.

docs/monitoring.md

@@ -1,6 +1,6 @@
-# Monitoring the NATS-Kafka Bridge
+# Monitoring the NATS-Replicator
 
-The nats-kafka bridge provides optional HTTP/s monitoring. When [configured with a monitoring port](config.md#monitoring) the server will provide two HTTP endpoints:
+The nats-replicator provides optional HTTP/s monitoring. When [configured with a monitoring port](config.md#monitoring) the server will provide two HTTP endpoints:
 
 * [/varz](#varz)
 * [/healthz](#healthz)
@@ -11,9 +11,9 @@ The nats-kafka bridge provides optional HTTP/s monitoring. When [configured with
 
 The `/varz` endpoint returns a JSON encoded set of statistics for the server. These statistics are wrapped in a root level object with the following properties:
 
-* `start_time` - the start time of the bridge, in the bridge's timezone.
-* `current_time` - the current time, in the bridge's timezone.
-* `uptime` - a string representation of the server's up time.
+* `start_time` - the start time of the replicator, in the replicator's timezone.
+* `current_time` - the current time, in the replicator's timezone.
+* `uptime` - a string representation of the replicator's up time.
 * `http_requests` - a map of request paths to counts, the keys are `/`, `/varz` and `/healthz`.
 * `connectors` - an array of statistics for each connector.
 
@@ -38,4 +38,4 @@ Each object in the connectors array, one per connector, will contain the followi
 
 ## /healthz
 
-The `/healthz` endpoint is provided for automated up/down style checks. The server returns an HTTP/200 when running and won't respond if it is down.
+The `/healthz` endpoint is provided for automated up/down style checks. The server returns an HTTP/200 when running and won't respond if it is down.

server/core/server_test.go

@@ -241,8 +241,8 @@ func (tbs *TestEnv) StartNATSandStan(port int, clusterID string, clientID string
 	return nil
 }
 
-// StopBridge stops the bridge
-func (tbs *TestEnv) StopBridge() {
+// StopReplicator stops the bridge
+func (tbs *TestEnv) StopReplicator() {
 	if tbs.Bridge != nil {
 		tbs.Bridge.Stop()
 		tbs.Bridge = nil

server/core/stan2nats.go

@@ -50,11 +50,11 @@ func (conn *Stan2NATSConnector) Start() error {
 		return fmt.Errorf("%s connector is improperly configured, incoming and outgoing settings are required", conn.String())
 	}
 
-	if !conn.bridge.CheckNATS(incoming) {
+	if !conn.bridge.CheckStan(incoming) {
 		return fmt.Errorf("%s connector requires nats connection named %s to be available", conn.String(), incoming)
 	}
 
-	if !conn.bridge.CheckStan(outgoing) {
+	if !conn.bridge.CheckNATS(outgoing) {
 		return fmt.Errorf("%s connector requires stan connection named %s to be available", conn.String(), outgoing)
 	}
 
@@ -127,7 +127,11 @@ func (conn *Stan2NATSConnector) Start() error {
 	conn.sub = sub
 
 	conn.stats.AddConnect()
-	conn.bridge.Logger().Tracef("opened and reading %s", conn.config.IncomingSubject)
+	if config.IncomingDurableName != "" {
+		conn.bridge.Logger().Tracef("opened and reading %s with durable name %s", conn.config.IncomingChannel, config.IncomingDurableName)
+	} else {
+		conn.bridge.Logger().Tracef("opened and reading %s", conn.config.IncomingChannel)
+	}
 	conn.bridge.Logger().Noticef("started connection %s", conn.String())
 
 	return nil
@@ -145,7 +149,7 @@ func (conn *Stan2NATSConnector) Shutdown() error {
 	conn.sub = nil
 
 	if sub != nil {
-		if err := sub.Unsubscribe(); err != nil {
+		if err := sub.Close(); err != nil {
 			conn.bridge.Logger().Noticef("error unsubscribing for %s, %s", conn.String(), err.Error())
 		}
 	}