package plugisservice import ( "context" "encoding/json" "errors" "iter" "log/slog" "os" "runtime" "time" "github.com/telemac/goutils/net" "github.com/telemac/goutils/task" "github.com/go-viper/mapstructure/v2" "github.com/nats-io/nats.go" "github.com/synadia-io/orbit.go/natsext" ) // Ensure Plugis implements PlugisIntf var _ PlugisIntf = (*Plugis)(nil) // Plugis is the default implementation of the PlugisIntf that provides // the functionality to plugis services. type Plugis struct { logger *slog.Logger nc *nats.Conn } var ( // ErrNatsNotConnected = errors.New("nats not connected") ErrNatsConnectionNil = errors.New("nats connection is nil") ) // SetLogger sets the logger for the Plugis instance. func (plugis *Plugis) SetLogger(log *slog.Logger) { plugis.logger = log } // Logger returns the logger for the Plugis instance. func (plugis *Plugis) Logger() *slog.Logger { if plugis.logger == nil { plugis.logger = slog.Default() } return plugis.logger } // SetNats sets the nats connection for the Plugis instance. func (plugis *Plugis) SetNats(nc *nats.Conn) { plugis.nc = nc } // Nats returns the nats connection for the Plugis instance. func (plugis *Plugis) Nats() *nats.Conn { return plugis.nc } // Publish publishes a message to the nats connection. func (plugis *Plugis) Publish(topic string, payload []byte) error { attrs := []slog.Attr{ slog.String("fn", "Plugis.Publish"), slog.String("topic", topic), slog.String("payload", string(payload)), } if plugis.nc == nil { return ErrNatsConnectionNil } err := plugis.nc.Publish(topic, payload) if err != nil { attrs = append(attrs, slog.String("err", err.Error())) plugis.Logger().LogAttrs(context.TODO(), slog.LevelError, "Publish payload to topic", attrs...) } else { plugis.Logger().LogAttrs(context.TODO(), slog.LevelDebug, "Published payload to topic", attrs...) } return err } // Prefix returns the prefix for the Plugis instance. func (plugis *Plugis) Prefix() string { return "integrator.customer.location" } // Request sends a request to the nats connection. func (plugis *Plugis) Request(subj string, data []byte, timeout time.Duration) (*nats.Msg, error) { attrs := []slog.Attr{ slog.String("fn", "Plugis.Request"), slog.String("subject", subj), slog.String("data", string(data)), slog.Duration("timeout", timeout), } if plugis.nc == nil { attrs = append(attrs, slog.String("err", ErrNatsConnectionNil.Error())) plugis.Logger().LogAttrs(context.TODO(), slog.LevelError, "Request failed - nats connection is nil", attrs...) return nil, ErrNatsConnectionNil } msg, err := plugis.nc.Request(subj, data, timeout) if err != nil { attrs = append(attrs, slog.String("err", err.Error())) plugis.Logger().LogAttrs(context.TODO(), slog.LevelError, "Request failed", attrs...) } else { plugis.Logger().LogAttrs(context.TODO(), slog.LevelDebug, "Request successful", attrs...) } return msg, err } // RequestCtx sends a request to the nats connection and returns a single message. // context is used for timeout and cancellation. func (plugis *Plugis) RequestCtx(ctx context.Context, subj string, data []byte) (*nats.Msg, error) { return nil, errors.New("not implemented") deadline, ok := ctx.Deadline() if !ok { //ctx, _ = context.WithTimeout(ctx, time.Hour*24) } else { _ = deadline } iter, err := plugis.RequestMany(ctx, subj, data, natsext.RequestManyMaxMessages(1)) plugis.Logger().Warn("RequestMany", slog.String("subject", subj), "err", err, "ctxErr", ctx.Err(), "cancealed", task.IsCancelled(ctx), ) if err != nil { return nil, err } if ctx.Err() != nil { return nil, ctx.Err() } for msg := range iter { return msg, nil } return &nats.Msg{}, ctx.Err() } // RequestMany sends a request to the nats connection and returns a sequence of messages. func (plugis *Plugis) RequestMany(ctx context.Context, subject string, data []byte, opts ...natsext.RequestManyOpt) (iter.Seq2[*nats.Msg, error], error) { attrs := []slog.Attr{ slog.String("fn", "Plugis.RequestMany"), slog.String("subject", subject), slog.String("data", string(data)), } if plugis.nc == nil { attrs = append(attrs, slog.String("err", ErrNatsConnectionNil.Error())) plugis.Logger().LogAttrs(context.TODO(), slog.LevelError, "RequestMany failed - nats connection is nil", attrs...) return nil, ErrNatsConnectionNil } result, err := natsext.RequestMany(ctx, plugis.nc, subject, data, opts...) if err != nil { attrs = append(attrs, slog.String("err", err.Error())) plugis.Logger().LogAttrs(context.TODO(), slog.LevelError, "RequestMany failed", attrs...) } else { plugis.Logger().LogAttrs(context.TODO(), slog.LevelDebug, "RequestMany successful", attrs...) } return result, err } // GetServices sends a request to the $SRV.INFO subject and returns a list of services. func (plugis *Plugis) GetServices(ctx context.Context) ([]ServiceInfo, error) { iter, err := plugis.RequestMany(ctx, "$SRV.INFO", []byte("")) if err != nil { return nil, err } var services []ServiceInfo for msg := range iter { var serviceInfo ServiceInfo err := json.Unmarshal(msg.Data, &serviceInfo) if err != nil { return nil, err } services = append(services, serviceInfo) } return services, nil } // isRunningInDockerContainer func isRunningInDockerContainer() bool { if _, err := os.Stat("/.dockerenv"); err == nil { return true } return false } // NewServiceMetadata creates and fills a ServiceMetadata structure func NewServiceMetadata(prefix string, startedAt time.Time) (Metadata, error) { var err error var meta ServiceMetadata meta.Prefix = prefix meta.Platform = runtime.GOOS + "/" + runtime.GOARCH if isRunningInDockerContainer() { meta.Platform += " (docker)" } meta.StartedAt = startedAt.String() meta.Hostname, err = os.Hostname() if err != nil { return nil, err } meta.MAC, err = net.GetMACAddress() if err != nil { return nil, err } // Convert struct to map using mapstructure var result Metadata err = mapstructure.Decode(meta, &result) return result, err }