chore: do not try to update stream to have the same behaviour as with…

…out create_if_not_exists flag.

internal/impl/nats/input_jetstream.go

@@ -19,7 +19,6 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
-	"strings"
 	"sync"
 	"time"
 
@@ -383,23 +382,6 @@ func (j *jetStreamReader) Connect(ctx context.Context) (err error) {
 						Replicas: j.numReplicas,
 					})
 				}
-			} else if strings.Contains(err.Error(), "does not match consumer") {
-				// create subject on existent stream
-				_, err = jCtx.UpdateStream(&nats.StreamConfig{
-					Name: j.stream,
-					Subjects: func() []string {
-						if j.subject == "" {
-							return nil
-						}
-						return []string{j.subject}
-					}(),
-					Storage: func() nats.StorageType {
-						if j.storageType == "file" {
-							return nats.FileStorage
-						}
-						return nats.MemoryStorage
-					}(),
-				})
 			}
 		}
 		return err
@@ -410,6 +392,18 @@ func (j *jetStreamReader) Connect(ctx context.Context) (err error) {
 	return nil
 }
 
+func unique(input []string) []string {
+	uniqueMap := make(map[string]struct{})
+	var result []string
+	for _, item := range input {
+		if _, exists := uniqueMap[item]; !exists {
+			uniqueMap[item] = struct{}{}
+			result = append(result, item)
+		}
+	}
+	return result
+}
+
 func (j *jetStreamReader) disconnect() {
 	j.connMut.Lock()
 	defer j.connMut.Unlock()