-
Notifications
You must be signed in to change notification settings - Fork 3
/
main.go
153 lines (132 loc) · 4.21 KB
/
main.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
//go:generate wit-bindgen-wrpc go --out-dir bindings --package github.com/couchbase-examples/wasmcloud-provider-couchbase/bindings wit
package main
import (
"context"
"errors"
"log"
"os"
"os/signal"
"syscall"
"time"
server "github.com/couchbase-examples/wasmcloud-provider-couchbase/bindings"
"github.com/couchbase/gocb/v2"
"github.com/wasmCloud/provider-sdk-go"
)
func main() {
if err := run(); err != nil {
log.Fatal(err)
}
}
func run() error {
// Handle SIGINT (CTRL+C) gracefully.
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
defer stop()
// Set up OpenTelemetry.
otelShutdown, err := setupOTelSDK(ctx)
if err != nil {
return err
}
// Handle shutdown properly so nothing leaks.
defer func() {
err = errors.Join(err, otelShutdown(context.Background()))
}()
// Initialize the provider with callbacks to track linked components
providerHandler := Handler{
linkedFrom: make(map[string]map[string]string),
clusterConnections: make(map[string]*gocb.Collection),
}
p, err := provider.New(
provider.TargetLinkPut(func(link provider.InterfaceLinkDefinition) error {
return handleNewTargetLink(&providerHandler, link)
}),
provider.TargetLinkDel(func(link provider.InterfaceLinkDefinition) error {
return handleDelTargetLink(&providerHandler, link)
}),
provider.HealthCheck(func() string {
return handleHealthCheck(&providerHandler)
}),
provider.Shutdown(func() error {
return handleShutdown(&providerHandler)
}),
)
if err != nil {
return err
}
// Store the provider for use in the handlers
providerHandler.WasmcloudProvider = p
// Setup two channels to await RPC and control interface operations
providerCh := make(chan error, 1)
signalCh := make(chan os.Signal, 1)
// Handle RPC operations
stopFunc, err := server.Serve(p.RPCClient, &providerHandler)
if err != nil {
p.Shutdown()
return err
}
// Handle control interface operations
go func() {
err := p.Start()
providerCh <- err
}()
// Shutdown on SIGINT
signal.Notify(signalCh, syscall.SIGINT)
// Run provider until either a shutdown is requested or a SIGINT is received
select {
case err = <-providerCh:
stopFunc()
return err
case <-signalCh:
p.Shutdown()
stopFunc()
}
return nil
}
// Provider handler functions
func handleNewTargetLink(handler *Handler, link provider.InterfaceLinkDefinition) error {
handler.Logger.Info("Handling new target link", "link", link)
handler.linkedFrom[link.SourceID] = link.TargetConfig
couchbaseConnectionArgs, err := validateCouchbaseConfig(link.TargetConfig, link.TargetSecrets)
if err != nil {
handler.Logger.Error("Invalid couchbase target config", "error", err)
return err
}
handler.updateCouchbaseCluster(handler, link.SourceID, couchbaseConnectionArgs)
return nil
}
func (h *Handler) updateCouchbaseCluster(handler *Handler, sourceId string, connectionArgs CouchbaseConnectionArgs) {
// Connect to the cluster
cluster, err := gocb.Connect(connectionArgs.ConnectionString, gocb.ClusterOptions{
Username: connectionArgs.Username,
Password: connectionArgs.Password,
})
if err != nil {
handler.Logger.Error("unable to connect to couchbase cluster", "error", err)
return
}
var collection *gocb.Collection
if connectionArgs.CollectionName != "" && connectionArgs.ScopeName != "" {
collection = cluster.Bucket(connectionArgs.BucketName).Scope(connectionArgs.ScopeName).Collection(connectionArgs.CollectionName)
} else {
collection = cluster.Bucket(connectionArgs.BucketName).DefaultCollection()
}
bucket := cluster.Bucket(connectionArgs.BucketName)
if err = bucket.WaitUntilReady(5*time.Second, nil); err != nil {
handler.Logger.Error("unable to connect to couchbase bucket", "error", err)
}
// Store the connection
handler.clusterConnections[sourceId] = collection
}
func handleDelTargetLink(handler *Handler, link provider.InterfaceLinkDefinition) error {
handler.Logger.Info("Handling del target link", "link", link)
delete(handler.linkedFrom, link.Target)
return nil
}
func handleHealthCheck(handler *Handler) string {
handler.Logger.Debug("Handling health check")
return "provider healthy"
}
func handleShutdown(handler *Handler) error {
handler.Logger.Info("Handling shutdown")
// clear(handler.linkedFrom)
return nil
}