jetstream.go

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/nats-io/nats.go"
)

const streamName = "foo"
const subjectName = "foo"

func main() {
	nc, err := nats.Connect(nats.DefaultURL)
	if err != nil {
		panic(err)
	}
	js, err := nc.JetStream()
	if err != nil {
		panic(err)
	}
	_, err = js.AddStream(&nats.StreamConfig{
		Name:      streamName,
		Subjects:  []string{subjectName},
		Retention: nats.InterestPolicy,
	})
	if err != nil {
		panic(err)
	}

	_, err = js.Subscribe(subjectName, func(msg *nats.Msg) {
		log.Println("C1_receive:", string(msg.Data)) // not Ack
		//log.Println("11111_replay",string(msg.Data))
		//if err := msg.Ack(); err != nil {
		//	log.Fatal("ack", err)
		//}

	}, nats.Durable("C1"), nats.ManualAck())
	if err != nil {
		panic(err)
	}

	_, err = js.Subscribe(subjectName, func(msg *nats.Msg) {
		log.Println("C2_receive:", string(msg.Data))
		// log.Println("2222_replay:", string(msg.Data)) // not ack
		//if err := msg.Ack(); err != nil {
		//	log.Fatal("ack", err)
		//}
	}, nats.Durable("C2"), nats.ManualAck())

	if err != nil {
		panic(err)
	}

	go func() {
		i := 0
		ticker := time.NewTicker(1 * time.Second)
		for range ticker.C {
			i++
			log.Println("publish ", i)
			ack, err := js.Publish(subjectName, []byte(fmt.Sprintf("hello %d", i)))
			if err != nil {
				panic(err)
			}
			log.Printf("%+v\n", ack)
		}
	}()

	<-time.After(1 * time.Hour)
}

// Copyright 2012-2019 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package main

//go:generate go run server/errors_gen.go

import (
"encoding/json"
"flag"
"fmt"
"log"
"os"

"github.com/nats-io/nats-server/v2/server"

)

var usageStr = `
Usage: nats-server [options]

Server Options:
-a, --addr Bind to host address (default: 0.0.0.0)
-p, --port Use port for clients (default: 4222)
-n, --name <server_name> Server name (default: auto)
-P, --pid File to store PID
-m, --http_port Use port for http monitoring
-ms,--https_port Use port for https monitoring
-c, --config Configuration file
-t Test configuration and exit
-sl,--signal [=] Send signal to nats-server process (stop, quit, reopen, reload)
can be either a PID (e.g. 1) or the path to a PID file (e.g. /var/run/nats-server.pid)
--client_advertise Client URL to advertise to other servers

Logging Options:
-l, --log File to redirect log output
-T, --logtime Timestamp log entries (default: true)
-s, --syslog Log to syslog or windows event log
-r, --remote_syslog Syslog server addr (udp://localhost:514)
-D, --debug Enable debugging output
-V, --trace Trace the raw protocol
-VV Verbose trace (traces system account as well)
-DV Debug and trace
-DVV Debug and verbose trace (traces system account as well)

JetStream Options:
-js, --jetstream Enable JetStream functionality.
-sd, --store_dir

Set the storage directory.

Authorization Options:
--user User required for connections
--pass Password required for connections
--auth Authorization token required for connections

TLS Options:
--tls Enable TLS, do not verify clients (default: false)
--tlscert Server certificate file
--tlskey Private key for server certificate
--tlsverify Enable TLS, verify client certificates
--tlscacert Client certificate CA for verification

Cluster Options:
--routes <rurl-1, rurl-2> Routes to solicit and connect
--cluster Cluster URL for solicited routes
--cluster_name Cluster Name, if not set one will be dynamically generated
--no_advertise Do not advertise known cluster information to clients
--cluster_advertise Cluster URL to advertise to other servers
--connect_retries For implicit routes, number of connect retries

Common Options:
-h, --help Show this message
-v, --version Show version
--help_tls TLS help
`

// usage will print out the flag options for the server.
func usage() {
fmt.Printf("%s\n", usageStr)
os.Exit(0)
}

func main() {
exe := "nats-server"

// Create a FlagSet and sets the usage
fs := flag.NewFlagSet(exe, flag.ExitOnError)
fs.Usage = usage

// Configure the options from the flags/config file
opts, err := server.ConfigureOptions(fs, os.Args[1:],
	server.PrintServerAndExit,
	fs.Usage,
	server.PrintTLSHelpAndDie)
if err != nil {
	server.PrintAndDie(fmt.Sprintf("%s: %s", exe, err))
} else if opts.CheckConfig {
	fmt.Fprintf(os.Stderr, "%s: configuration file %s is valid\n", exe, opts.ConfigFile)
	os.Exit(0)
}
bt, _ :=json.Marshal(opts)
log.Println(string(bt))
opts.Trace=true
opts.Port=4222
opts.JetStream=true
// Create the server with appropriate options.
s, err := server.NewServer(opts)
if err != nil {
	server.PrintAndDie(fmt.Sprintf("%s: %s", exe, err))
}

// Configure the logger based on the flags
s.ConfigureLogger()

// Start things up. Block here until done.
if err := server.Run(s); err != nil {
	server.PrintAndDie(err.Error())
}
s.WaitForShutdown()

}