Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update NATS-Streaming dependencies version (#1533)

Browse Source
This commit is contained in:
Ta-Ching Chen
2020-02-12 20:35:20 +08:00
committed by GitHub
parent f2ddaa7921
commit 735f9abef7
11 changed files with 404 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/tests/mqtrigger/nats/stan-pub.go → test/tests/mqtrigger/nats/stan-pub/main.go
+46 -15
View File
@@ -1,9 +1,19 @@
// This file originally came from official Nats.io GitHub repository.
// You can reach original file with the following link:
// https://github.com/nats-io/go-nats-streaming/tree/master/examples
// You can reach the original file with the following link:
// https://github.com/nats-io/stan.go/blob/master/examples/stan-pub/main.go
// Copyright 2012-2016 Apcera Inc. All rights reserved.
// +build ignore
// Copyright 2016-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
@@ -15,17 +25,19 @@ import (
"sync"
"time"
"github.com/nats-io/go-nats-streaming"
nats "github.com/nats-io/nats.go"
"github.com/nats-io/stan.go"
)
var usageStr = `
Usage: stan-pub [options] <subject> <message>
Options:
-s, --server <url> NATS Streaming server URL(s)
-c, --cluster <cluster name> NATS Streaming cluster name
-id,--clientid <client ID> NATS Streaming client ID
-a, --async Asynchronous publish mode
-s, --server <url> NATS Streaming server URL(s)
-c, --cluster <cluster name> NATS Streaming cluster name
-id, --clientid <client ID> NATS Streaming client ID
-a, --async Asynchronous publish mode
-cr, --creds <credentials> NATS 2.0 Credentials
`
// NOTE: Use tls scheme for TLS, e.g. stan-pub -s tls://demo.nats.io:4443 foo hello
@@ -35,10 +47,13 @@ func usage() {
}
func main() {
var clusterID string
var clientID string
var async bool
var URL string
var (
clusterID string
clientID string
URL string
async bool
userCreds string
)
flag.StringVar(&URL, "s", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
flag.StringVar(&URL, "server", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
@@ -48,6 +63,8 @@ func main() {
flag.StringVar(&clientID, "clientid", "stan-pub", "The NATS Streaming client ID to connect with")
flag.BoolVar(&async, "a", false, "Publish asynchronously")
flag.BoolVar(&async, "async", false, "Publish asynchronously")
flag.StringVar(&userCreds, "cr", "", "Credentials File")
flag.StringVar(&userCreds, "creds", "", "Credentials File")
log.SetFlags(0)
flag.Usage = usage
@@ -59,7 +76,21 @@ func main() {
usage()
}
sc, err := stan.Connect(clusterID, clientID, stan.NatsURL(URL))
// Connect Options.
opts := []nats.Option{nats.Name("NATS Streaming Example Publisher")}
// Use UserCredentials
if userCreds != "" {
opts = append(opts, nats.UserCredentials(userCreds))
}
// Connect to NATS
nc, err := nats.Connect(URL, opts...)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
sc, err := stan.Connect(clusterID, clientID, stan.NatsConn(nc))
if err != nil {
log.Fatalf("Can't connect: %v.\nMake sure a NATS Streaming Server is running at: %s", err, URL)
}
@@ -83,7 +114,7 @@ func main() {
ch <- true
}
if async != true {
if !async {
err = sc.Publish(subj, msg)
if err != nil {
log.Fatalf("Error during publish: %v\n", err)
test/tests/mqtrigger/nats/stan-sub.go
-133
View File
@@ -1,133 +0,0 @@
// This file originally came from official Nats.io GitHub repository.
// You can reach original file with the following link:
// https://github.com/nats-io/go-nats-streaming/tree/master/examples
// Copyright 2012-2016 Apcera Inc. All rights reserved.
// +build ignore
package main
import (
"flag"
"log"
"time"
"github.com/nats-io/go-nats-streaming"
"github.com/nats-io/go-nats-streaming/pb"
)
var usageStr = `
Usage: stan-sub [options] <subject>
Options:
-s, --server <url> NATS Streaming server URL(s)
-c, --cluster <cluster name> NATS Streaming cluster name
-id,--clientid <client ID> NATS Streaming client ID
Subscription Options:
--qgroup <name> Queue group
--seq <seqno> Start at seqno
--all Deliver all available messages
--last Deliver starting with last published message
--since <duration> Deliver messages in last interval (e.g. 1s, 1hr)
(for more information: https://golang.org/pkg/time/#ParseDuration)
--durable <name> Durable subscriber name
--unsubscribe Unsubscribe the durable on exit
`
// NOTE: Use tls scheme for TLS, e.g. stan-sub -s tls://demo.nats.io:4443 foo
func usage() {
log.Fatalf(usageStr)
}
func printMsg(m *stan.Msg) {
log.Printf("[%s]: '%s'", m.Subject, m.Data)
}
func main() {
var clusterID string
var clientID string
var showTime bool
var startSeq uint64
var startDelta string
var deliverAll bool
var deliverLast bool
var durable string
var qgroup string
var unsubscribe bool
var URL string
// defaultID := fmt.Sprintf("client.%s", nuid.Next())
flag.StringVar(&URL, "s", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
flag.StringVar(&URL, "server", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
flag.StringVar(&clusterID, "c", "test-cluster", "The NATS Streaming cluster ID")
flag.StringVar(&clusterID, "cluster", "test-cluster", "The NATS Streaming cluster ID")
flag.StringVar(&clientID, "id", "", "The NATS Streaming client ID to connect with")
flag.StringVar(&clientID, "clientid", "", "The NATS Streaming client ID to connect with")
flag.BoolVar(&showTime, "t", false, "Display timestamps")
// Subscription options
flag.Uint64Var(&startSeq, "seq", 0, "Start at sequence no.")
flag.BoolVar(&deliverAll, "all", false, "Deliver all")
flag.BoolVar(&deliverLast, "last", false, "Start with last value")
flag.StringVar(&startDelta, "since", "", "Deliver messages since specified time offset")
flag.StringVar(&durable, "durable", "", "Durable subscriber name")
flag.StringVar(&qgroup, "qgroup", "", "Queue group name")
flag.BoolVar(&unsubscribe, "unsubscribe", false, "Unsubscribe the durable on exit")
log.SetFlags(0)
flag.Usage = usage
flag.Parse()
args := flag.Args()
if clientID == "" {
log.Printf("Error: A unique client ID must be specified.")
usage()
}
if len(args) < 1 {
log.Printf("Error: A subject must be specified.")
usage()
}
sc, err := stan.Connect(clusterID, clientID, stan.NatsURL(URL))
if err != nil {
log.Fatalf("Can't connect: %v.\nMake sure a NATS Streaming Server is running at: %s", err, URL)
}
// log.Printf("Connected to %s clusterID: [%s] clientID: [%s]\n", URL, clusterID, clientID)
subj := args[0]
exit := make(chan struct{})
mcb := func(msg *stan.Msg) {
printMsg(msg)
exit <- struct{}{}
}
startOpt := stan.StartAt(pb.StartPosition_NewOnly)
if startSeq != 0 {
startOpt = stan.StartAtSequence(startSeq)
} else if deliverLast == true {
startOpt = stan.StartWithLastReceived()
} else if deliverAll == true {
log.Print("subscribing with DeliverAllAvailable")
startOpt = stan.DeliverAllAvailable()
} else if startDelta != "" {
ago, err := time.ParseDuration(startDelta)
if err != nil {
sc.Close()
log.Fatal(err)
}
startOpt = stan.StartAtTimeDelta(ago)
}
sub, err := sc.QueueSubscribe(subj, qgroup, mcb, startOpt, stan.DurableName(durable))
if err != nil {
sc.Close()
log.Fatal(err)
}
<-exit
sub.Unsubscribe()
}
test/tests/mqtrigger/nats/stan-sub/main.go
+184
View File
@@ -0,0 +1,184 @@
// This file originally came from official Nats.io GitHub repository.
// You can reach the original file with the following link:
// https://github.com/nats-io/stan.go/blob/master/examples/stan-sub/main.go
// Copyright 2016-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
import (
"flag"
"fmt"
"log"
"os"
"os/signal"
"time"
nats "github.com/nats-io/nats.go"
"github.com/nats-io/stan.go"
"github.com/nats-io/stan.go/pb"
)
var usageStr = `
Usage: stan-sub [options] <subject>
Options:
-s, --server <url> NATS Streaming server URL(s)
-c, --cluster <cluster name> NATS Streaming cluster name
-id, --clientid <client ID> NATS Streaming client ID
-cr, --creds <credentials> NATS 2.0 Credentials
Subscription Options:
--qgroup <name> Queue group
--all Deliver all available messages
--last Deliver starting with last published message
--since <time_ago> Deliver messages in last interval (e.g. 1s, 1hr)
--seq <seqno> Start at seqno
--new_only Only deliver new messages
--durable <name> Durable subscriber name
--unsub Unsubscribe the durable on exit
`
// NOTE: Use tls scheme for TLS, e.g. stan-sub -s tls://demo.nats.io:4443 foo
func usage() {
log.Fatalf(usageStr)
}
func printMsg(m *stan.Msg, i int) {
log.Printf("[#%d] Received: %s\n", i, m)
}
func main() {
var (
clusterID, clientID string
URL string
userCreds string
showTime bool
qgroup string
unsubscribe bool
startSeq uint64
startDelta string
deliverAll bool
newOnly bool
deliverLast bool
durable string
)
flag.StringVar(&URL, "s", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
flag.StringVar(&URL, "server", stan.DefaultNatsURL, "The nats server URLs (separated by comma)")
flag.StringVar(&clusterID, "c", "test-cluster", "The NATS Streaming cluster ID")
flag.StringVar(&clusterID, "cluster", "test-cluster", "The NATS Streaming cluster ID")
flag.StringVar(&clientID, "id", "stan-sub", "The NATS Streaming client ID to connect with")
flag.StringVar(&clientID, "clientid", "stan-sub", "The NATS Streaming client ID to connect with")
flag.BoolVar(&showTime, "t", false, "Display timestamps")
// Subscription options
flag.Uint64Var(&startSeq, "seq", 0, "Start at sequence no.")
flag.BoolVar(&deliverAll, "all", true, "Deliver all")
flag.BoolVar(&newOnly, "new_only", false, "Only new messages")
flag.BoolVar(&deliverLast, "last", false, "Start with last value")
flag.StringVar(&startDelta, "since", "", "Deliver messages since specified time offset")
flag.StringVar(&durable, "durable", "", "Durable subscriber name")
flag.StringVar(&qgroup, "qgroup", "", "Queue group name")
flag.BoolVar(&unsubscribe, "unsub", false, "Unsubscribe the durable on exit")
flag.BoolVar(&unsubscribe, "unsubscribe", false, "Unsubscribe the durable on exit")
flag.StringVar(&userCreds, "cr", "", "Credentials File")
flag.StringVar(&userCreds, "creds", "", "Credentials File")
log.SetFlags(0)
flag.Usage = usage
flag.Parse()
args := flag.Args()
if len(args) < 1 {
log.Printf("Error: A subject must be specified.")
usage()
}
// Connect Options.
opts := []nats.Option{nats.Name("NATS Streaming Example Subscriber")}
// Use UserCredentials
if userCreds != "" {
opts = append(opts, nats.UserCredentials(userCreds))
}
// Connect to NATS
nc, err := nats.Connect(URL, opts...)
if err != nil {
log.Fatal(err)
}
defer nc.Close()
sc, err := stan.Connect(clusterID, clientID, stan.NatsConn(nc),
stan.SetConnectionLostHandler(func(_ stan.Conn, reason error) {
log.Fatalf("Connection lost, reason: %v", reason)
}))
if err != nil {
log.Fatalf("Can't connect: %v.\nMake sure a NATS Streaming Server is running at: %s", err, URL)
}
log.Printf("Connected to %s clusterID: [%s] clientID: [%s]\n", URL, clusterID, clientID)
// Process Subscriber Options.
startOpt := stan.StartAt(pb.StartPosition_NewOnly)
if startSeq != 0 {
startOpt = stan.StartAtSequence(startSeq)
} else if deliverLast {
startOpt = stan.StartWithLastReceived()
} else if deliverAll && !newOnly {
startOpt = stan.DeliverAllAvailable()
} else if startDelta != "" {
ago, err := time.ParseDuration(startDelta)
if err != nil {
sc.Close()
log.Fatal(err)
}
startOpt = stan.StartAtTimeDelta(ago)
}
subj, i := args[0], 0
mcb := func(msg *stan.Msg) {
i++
printMsg(msg, i)
}
sub, err := sc.QueueSubscribe(subj, qgroup, mcb, startOpt, stan.DurableName(durable))
if err != nil {
sc.Close()
log.Fatal(err)
}
log.Printf("Listening on [%s], clientID=[%s], qgroup=[%s] durable=[%s]\n", subj, clientID, qgroup, durable)
if showTime {
log.SetFlags(log.LstdFlags)
}
// Wait for a SIGINT (perhaps triggered by user with CTRL-C)
// Run cleanup when signal is received
signalChan := make(chan os.Signal, 1)
cleanupDone := make(chan bool)
signal.Notify(signalChan, os.Interrupt)
go func() {
for range signalChan {
fmt.Printf("\nReceived an interrupt, unsubscribing and closing connection...\n\n")
// Do not unsubscribe a durable on exit, except if asked to.
if durable == "" || unsubscribe {
sub.Unsubscribe()
}
sc.Close()
cleanupDone <- true
}
}()
<-cleanupDone
}
test/tests/mqtrigger/nats/test_mqtrigger.sh
+6 -9
View File
@@ -19,7 +19,8 @@ pubClientID="clientPub-$TEST_ID"
subClientID="clientSub-$TEST_ID"
topic="foo.bar$TEST_ID"
resptopic="foo.foo$TEST_ID"
expectedRespOutput="[foo.foo$TEST_ID]: 'Hello, World!'"
#FISSION_NATS_STREAMING_URL="http://defaultFissionAuthToken@$(minikube ip):4222"
expectedRespOutput="subject:\"$resptopic\" data:\"Hello, World!\""
env=nodejs-$TEST_ID
fn=hello-$TEST_ID
@@ -52,20 +53,16 @@ sleep 5
# Send a message
#
log "Sending message"
go run $DIR/stan-pub.go -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $pubClientID $topic ""
go run $DIR/stan-pub/main.go -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $pubClientID $topic ""
#
# Wait for message on response topic
#
log "Waiting for response"
response=$(timeout 120s go run $DIR/stan-sub.go --last -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $subClientID $resptopic 2>&1)
response=$(timeout 5s go run $DIR/stan-sub/main.go --last -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $subClientID $resptopic 2>&1 || true)
echo "$response" | grep "$expectedRespOutput"
if [[ "$response" != "$expectedRespOutput" ]]; then
log "'$response' is not equal to '$expectedRespOutput'"
exit 1
fi
log "Deleting message queue trigger"
log "Deleting message queue trigger"
fission mqtrigger delete --name $mqt
log "Subscriber received expected response: $response"
test/tests/mqtrigger/nats/test_mqtrigger_error.sh
+5 -13
View File
@@ -21,8 +21,8 @@ topic="foo.bar$TEST_ID"
resptopic="foo.foo$TEST_ID"
errortopic="foo.error$TEST_ID"
maxretries=1
# FISSION_NATS_STREAMING_URL="http://defaultFissionAuthToken@$(minikube ip):4222"
expectedRespOutput="[foo.error$TEST_ID]: 'Hello, World!'"
#FISSION_NATS_STREAMING_URL="http://defaultFissionAuthToken@$(minikube ip):4222"
expectedRespOutput="subject:\"$errortopic\" data:\"Hello, World!\""
env=nodejs-$TEST_ID
fn=hello-$TEST_ID
@@ -57,22 +57,14 @@ sleep 5
# Send a message
#
log "Sending message"
go run $DIR/stan-pub.go -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $pubClientID $topic ""
go run $DIR/stan-pub/main.go -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $pubClientID $topic ""
#
# Wait for message on error topic
#
log "Waiting for response"
response=$(go run $DIR/stan-sub.go --last -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $subClientID $errortopic 2>&1)
log "Subscriber received response: $response"
if [[ "$response" != "$expectedRespOutput" ]]; then
log "'$response' is not equal to '$expectedRespOutput'"
exit 1
else
log "Responses match."
fi
response=$(timeout 5s go run $DIR/stan-sub/main.go --last -s $FISSION_NATS_STREAMING_URL -c $clusterID -id $subClientID $errortopic 2>&1 || true)
echo "$response" | grep "$expectedRespOutput"
log "Deleting message queue trigger"
fission mqtrigger delete --name $mqt