Anthrove/plug-sdk
4
0
Fork 0
Code Issues Pull Requests Actions Projects Releases 6 Activity

refactor: move object functions to own ones

Browse Source
This commit is contained in:
Alphyron 2024-08-27 07:45:30 +02:00
parent c0f069b3c0
commit 23bae0a5f8
2 changed files with 19 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pkg/plug/grpc.go
View File

@ -61,7 +61,6 @@ func (s *server) TaskStart(ctx context.Context, creation *gRPC.PlugTaskCreation)
// by using this method we assign a new context to each new request we get. // by using this method we assign a new context to each new request we get.
// This can be used for example to close the context with the given id // This can be used for example to close the context with the given id
ctx = trace.ContextWithSpanContext(context.Background(), trace.NewSpanContext(trace.SpanContextConfig{TraceID: span.SpanContext().TraceID()})) ctx = trace.ContextWithSpanContext(context.Background(), trace.NewSpanContext(trace.SpanContextConfig{TraceID: span.SpanContext().TraceID()}))
taskCtx, cancel := context.WithCancel(ctx) taskCtx, cancel := context.WithCancel(ctx)
s.ctx[id] = cancel s.ctx[id] = cancel
span.AddEvent("Created new context for task", trace.WithAttributes(attribute.String("task_id", id))) span.AddEvent("Created new context for task", trace.WithAttributes(attribute.String("task_id", id)))

60
pkg/plug/server.go
View File

@ -2,7 +2,6 @@ package plug
import ( import (
"context" "context"
"fmt"
"net" "net"
"git.anthrove.art/Anthrove/otter-space-sdk/v3/pkg/database" "git.anthrove.art/Anthrove/otter-space-sdk/v3/pkg/database"
@ -28,55 +27,40 @@ type TaskExecution func(ctx context.Context, userSource models.UserSource, deepS
type SendMessageExecution func(ctx context.Context, userSource models.UserSource, message string) error type SendMessageExecution func(ctx context.Context, userSource models.UserSource, message string) error
type GetMessageExecution func(ctx context.Context, userSource models.UserSource) ([]Message, error) type GetMessageExecution func(ctx context.Context, userSource models.UserSource) ([]Message, error)
type Plug struct { var (
address string
port string
taskExecutionFunction TaskExecution taskExecutionFunction TaskExecution
sendMessageExecution SendMessageExecution sendMessageExecution SendMessageExecution
getMessageExecution GetMessageExecution getMessageExecution GetMessageExecution
source models.Source )
plugName string
}
func NewPlug(plugName string, address string, port string, source models.Source) Plug { func Listen(ctx context.Context, listenAddr string, source models.Source) error {
return Plug{
address: address,
port: port,
source: source,
plugName: plugName,
}
}
func (p *Plug) Listen(ctx context.Context) error {
ctx, span := tracer.Start(ctx, "Listen") ctx, span := tracer.Start(ctx, "Listen")
defer span.End() defer span.End()
var err error var err error
var source models.Source
span.SetAttributes( span.SetAttributes(
attribute.String("source_display_name", string(p.source.DisplayName)), attribute.String("source_display_name", string(source.DisplayName)),
attribute.String("source_domain", string(p.source.Domain)), attribute.String("source_domain", string(source.Domain)),
) )
sourceFields := log.Fields{ sourceFields := log.Fields{
"source_display_name": p.source.DisplayName, "source_display_name": source.DisplayName,
"source_domain": p.source.Domain, "source_domain": source.Domain,
} }
serverFields := log.Fields{ serverFields := log.Fields{
"address": p.address, "address": listenAddr,
"port": p.port,
} }
source, err = database.GetSourceByDomain(ctx, p.source.Domain) source, err = database.GetSourceByDomain(ctx, source.Domain)
if err != nil { if err != nil {
if err.Error() == "Database error: NoDataFound" { if err.Error() == "Database error: NoDataFound" {
span.AddEvent("No Source found, initializing source") span.AddEvent("No Source found, initializing source")
log.WithContext(ctx).WithFields(sourceFields).Info("No Source found, initializing source!") log.WithContext(ctx).WithFields(sourceFields).Info("No Source found, initializing source!")
source, err = database.CreateSource(ctx, p.source) source, err = database.CreateSource(ctx, source)
if err != nil { if err != nil {
span.RecordError(err) span.RecordError(err)
span.SetStatus(codes.Error, err.Error()) span.SetStatus(codes.Error, err.Error())
@ -95,13 +79,11 @@ func (p *Plug) Listen(ctx context.Context) error {
} }
} }
p.source = source
span.SetAttributes( span.SetAttributes(
attribute.String("source_id", string(p.source.ID)), attribute.String("source_id", string(source.ID)),
) )
lis, err := net.Listen("tcp", fmt.Sprintf("%s:%s", p.address, p.port)) lis, err := net.Listen("tcp", listenAddr)
if err != nil { if err != nil {
span.RecordError(err) span.RecordError(err)
span.SetStatus(codes.Error, err.Error()) span.SetStatus(codes.Error, err.Error())
@ -111,7 +93,7 @@ func (p *Plug) Listen(ctx context.Context) error {
grpcServer := grpc.NewServer() grpcServer := grpc.NewServer()
pb.RegisterPlugConnectorServer(grpcServer, NewGrpcServer(p.source, p.taskExecutionFunction, p.sendMessageExecution, p.getMessageExecution)) pb.RegisterPlugConnectorServer(grpcServer, NewGrpcServer(source, taskExecutionFunction, sendMessageExecution, getMessageExecution))
go func() { go func() {
err = grpcServer.Serve(lis) err = grpcServer.Serve(lis)
@ -131,18 +113,14 @@ func (p *Plug) Listen(ctx context.Context) error {
return nil return nil
} }
func (p *Plug) GetSource() models.Source { func SetTaskExecutionFunction(function TaskExecution) {
return p.source taskExecutionFunction = function
} }
func (p *Plug) TaskExecutionFunction(function TaskExecution) { func SetSendMessageExecutionFunction(function SendMessageExecution) {
p.taskExecutionFunction = function sendMessageExecution = function
} }
func (p *Plug) SendMessageExecution(function SendMessageExecution) { func SetGetMessageExecutionFunction(function GetMessageExecution) {
p.sendMessageExecution = function getMessageExecution = function
}
func (p *Plug) GetMessageExecution(function GetMessageExecution) {
p.getMessageExecution = function
} }