As stated above - the dispose fixed the issue. Process run stable the whole night. Thanks!
For completeness the working code:
package main
import (
“encoding/json”
“flag”
“fmt”
“GitHub - satori/go.uuid: UUID package for Go”
log “GitHub - sirupsen/logrus: Structured, pluggable logging for Go.”
“math/rand”
“os”
“os/signal”
“strconv”
“time”
“solace.dev/go/messaging”
“solace.dev/go/messaging/pkg/solace/config”
“solace.dev/go/messaging/pkg/solace/resource”
)
var (
logLevel = flag.Int(“logLevel”, 4, “log level (0-6)”)
logReportCaller = flag.Bool(“logReportCaller”, false, “add caller to log output”)
logFormatJson = flag.Bool(“logFormatJson”, true, “log in json format”)
)
func init() {
flag.Parse()
log.SetLevel(log.Level(*logLevel))
log.SetReportCaller(*logReportCaller)
log.SetFormatter(&log.TextFormatter{})
if *logFormatJson {
log.SetFormatter(&log.JSONFormatter{})
}
}
func getEnv(key string) string {
if val, ok := os.LookupEnv(key); ok {
return val
} else {
log.Info("Missing environment variable ", key)
os.Exit(8)
}
return “”
}
func main() {
p, err := strconv.Atoi(getEnv(“PARALLEL_PRODUCER”))
if err != nil {
log.Panic("PARALLEL_PRODUCER variable has to be a positive integer ")
}
msgSeqNum := 0
for i := 0; i < p; i++ {
go func(v int) {
msgLength, err := strconv.Atoi(getEnv(“MSG_LENGTH”))
if err != nil {
log.Panic(err)
}
TopicPrefix := getEnv(“SOLACE_TOPIC_PREFIX”)
log.Info(TopicPrefix)
// Configuration parameters
brokerConfig := config.ServicePropertyMap{
config.TransportLayerPropertyHost: getEnv(“SOLACE_HOST”),
config.ServicePropertyVPNName: getEnv(“SOLACE_VPN”),
config.AuthenticationPropertySchemeBasicUserName: getEnv(“SOLACE_USER”),
config.AuthenticationPropertySchemeBasicPassword: getEnv(“SOLACE_PASSWORD”),
}
messagingService, err := messaging.NewMessagingServiceBuilder().
FromConfigurationProvider(brokerConfig).
WithTransportSecurityStrategy(config.NewTransportSecurityStrategy().WithoutCertificateValidation()).
Build()
if err != nil {
panic(err)
}
// Connect to the messaging serice
if err := messagingService.Connect(); err != nil {
panic(err)
}
log.Info("Connected to the broker? ", messagingService.IsConnected())
// Build a Direct Message Publisher
directPublisher, builderErr := messagingService.CreateDirectMessagePublisherBuilder().Build()
if builderErr != nil {
panic(builderErr)
}
startErr := directPublisher.Start()
if startErr != nil {
panic(startErr)
}
log.Info(“Direct Publisher running? “, directPublisher.IsRunning())
for directPublisher.IsReady() {
// Prepare outbound message payload and body
var messageBody string
messageBody, err = getRandomMessage(msgLength)
if err != nil {
log.Panic(err)
}
id := uuid.NewV4().String()
messageBuilder := messagingService.MessageBuilder().
WithProperty(“application”, “samples”).
WithProperty(“language”, “go”)
msgSeqNum++
message, err := messageBuilder.BuildWithStringPayload(messageBody)
if err != nil {
panic(err)
}
topic := TopicPrefix + “/” + fmt.Sprintf(”%d”, v) + “/” + id
log.Info("topic: “, topic)
publishErr := directPublisher.Publish(message, resource.TopicOf(topic))
if publishErr != nil {
panic(publishErr)
}
log.Info(“message published:”, messageBody)
message.Dispose()
}
// Terminate the Direct Publisher
directPublisher.Terminate(2 * time.Second)
log.Info(”\nDirect Publisher Terminated? ", directPublisher.IsTerminated())
// Disconnect the Message Service
messagingService.Disconnect()
log.Info("Messaging Service Disconnected? ", !messagingService.IsConnected())
}(i)
}
// Handle interrupts
c := make(chan os.Signal, 1)
signal.Notify(c, os.Interrupt)
// Block until a signal is received.
<-c
// TODO
// Find way to shutdown the go routine
// e.g use another channel, BOOl…etc
// TODO
}
var letterBytes = “abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789”
func getRandomMessage(n int) (string, error) {
// Seed the random number generator
rand.Seed(time.Now().UnixNano())
// Generate a random string of length n
b := make(byte, n)
for i := range b {
b[i] = letterBytes[rand.Intn(len(letterBytes))]
}
randomStr := string(b)
// Generate a UUID
id := uuid.NewV4()
// Create the response object
response := struct {
Message stringjson:"message"
ID stringjson:"id"
}{
Message: randomStr,
ID: id.String(),
}
// Marshal the response to a JSON string
responseJSON, err := json.Marshal(response)
if err != nil {
return “”, err
}
return string(responseJSON), nil
}