1
2
3
4
5
6

Node (Cihaz/Sunucu)
  └── Device (Mantıksal İşlem Birimi)
       ├── Source (Medya Kaynağı)
       ├── Flow (Mantıksal Medya Akışı)
       ├── Sender (Flow'u İletir)
       └── Receiver (Flow'u Alır)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

nmos-registry/
├── main.go
├── registry/
│   ├── registry.go        # Temel registry mantığı
│   ├── registration.go    # Registration API (IS-04)
│   ├── query.go           # Query API (IS-04)
│   ├── websocket.go       # WebSocket subscription'lar
│   └── models.go          # NMOS kaynak modelleri
├── storage/
│   ├── memory.go          # In-memory depolama
│   └── postgres.go        # PostgreSQL depolama (production)
└── go.mod

  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

// registry/models.go
package registry

import (
    "time"
)

// Resource, tüm NMOS kaynakları için temel
type Resource struct {
    ID          string            `json:"id"`
    Version     string            `json:"version"`
    Label       string            `json:"label"`
    Description string            `json:"description"`
    Tags        map[string]string `json:"tags"`
}

// Node, cihazlar için mantıksal host'u temsil eder
type Node struct {
    Resource
    Hostname    string      `json:"hostname"`
    APIVersions []string    `json:"api"`
    Caps        interface{} `json:"caps"`
    Services    []Service   `json:"services"`
    Clocks      []Clock     `json:"clocks"`
    Interfaces  []Interface `json:"interfaces"`
}

// Device, bir işlevsellik birimini temsil eder (encoder, decoder, vb.)
type Device struct {
    Resource
    Type      string   `json:"type"` // generic, pipeline, host
    NodeID    string   `json:"node_id"`
    Senders   []string `json:"senders"`   // Sender ID'leri dizisi
    Receivers []string `json:"receivers"` // Receiver ID'leri dizisi
    Controls  []Control `json:"controls"`
}

// Source, medya içeriğinin kaynağını temsil eder
type Source struct {
    Resource
    Format    string      `json:"format"` // video, audio, data
    DeviceID  string      `json:"device_id"`
    ClockName string      `json:"clock_name"`
    GrainRate *Rational   `json:"grain_rate,omitempty"`
    // ... ek alanlar: caps, parents, ses için channels
}

// Flow, bir medya akışını temsil eder
type Flow struct {
    Resource
    Format       string    `json:"format"` // urn:x-nmos:format:video, audio, data
    SourceID     string    `json:"source_id"`
    DeviceID     string    `json:"device_id"`
    GrainRate    *Rational `json:"grain_rate,omitempty"`
    MediaType    string    `json:"media_type"` // video/raw, audio/L24, vb.
    
    // Video-spesifik: frame_width, frame_height, colorspace, interlace_mode
    // Ses-spesifik: sample_rate, bit_depth, channels
    // ... tam alan listesi için NMOS IS-04 spec'e bakın
}

// Sender bir flow'u iletir
type Sender struct {
    Resource
    FlowID          string         `json:"flow_id"`
    Transport       string         `json:"transport"` // urn:x-nmos:transport:rtp
    DeviceID        string         `json:"device_id"`
    ManifestHref    string         `json:"manifest_href"` // SDP dosya URL'i
    InterfaceBindings []string     `json:"interface_bindings"`
    SubscriptionID  string         `json:"subscription,omitempty"`
}

// Receiver bir flow'u alır
type Receiver struct {
    Resource
    Format       string         `json:"format"` // urn:x-nmos:format:video
    Caps         ReceiverCaps   `json:"caps"`
    Transport    string         `json:"transport"`
    DeviceID     string         `json:"device_id"`
    InterfaceBindings []string  `json:"interface_bindings"`
    SubscriptionID string       `json:"subscription,omitempty"`
}

// Destekleyici yapılar (basitleştirilmiş - tüm alanlar için spec'e bakın)
type Service struct {
    Href string `json:"href"`
    Type string `json:"type"`
}

type Clock struct {
    Name    string `json:"name"`
    RefType string `json:"ref_type"` // internal, ptp, vb.
}

type Interface struct {
    Name string `json:"name"`
    // ... chassis_id, port_id, vb.
}

type Rational struct {
    Numerator   int `json:"numerator"`
    Denominator int `json:"denominator"`
}

type ReceiverCaps struct {
    MediaTypes []string `json:"media_types,omitempty"`
}

// Tam kaynak tanımları için: https://specs.amwa.tv/is-04/

// Version, NMOS version zaman damgası oluşturur
func CreateVersion() string {
    now := time.Now()
    seconds := now.Unix()
    nanos := now.Nanosecond()
    return fmt.Sprintf("%d:%d", seconds, nanos)
}

  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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279

// registry/registry.go
package registry

import (
    "fmt"
    "sync"
    "time"
)

// Registry, NMOS kaynaklarını yönetir
type Registry struct {
    mu         sync.RWMutex
    nodes      map[string]*Node
    devices    map[string]*Device
    sources    map[string]*Source
    flows      map[string]*Flow
    senders    map[string]*Sender
    receivers  map[string]*Receiver
    
    // Sağlık takibi
    health     map[string]time.Time // node_id -> son heartbeat
    
    // WebSocket subscription'lar
    subscribers map[string]*Subscriber
}

type Subscriber struct {
    ID       string
    Query    SubscriptionQuery
    Updates  chan ResourceUpdate
}

type SubscriptionQuery struct {
    ResourceType string            // node, device, sender, receiver
    Params       map[string]string // sorgu parametreleri
}

type ResourceUpdate struct {
    Action   string      // create, update, delete
    Type     string      // kaynak türü
    Resource interface{} // kaynak
}

// NewRegistry yeni bir NMOS registry oluşturur
func NewRegistry() *Registry {
    r := &Registry{
        nodes:       make(map[string]*Node),
        devices:     make(map[string]*Device),
        sources:     make(map[string]*Source),
        flows:       make(map[string]*Flow),
        senders:     make(map[string]*Sender),
        receivers:   make(map[string]*Receiver),
        health:      make(map[string]time.Time),
        subscribers: make(map[string]*Subscriber),
    }
    
    // Sağlık kontrolü goroutine'ini başlat
    go r.healthCheckLoop()
    
    return r
}

// RegisterNode bir node'u kaydeder veya günceller
func (r *Registry) RegisterNode(node *Node) error {
    r.mu.Lock()
    defer r.mu.Unlock()
    
    existing, exists := r.nodes[node.ID]
    action := "create"
    
    if exists {
        action = "update"
        // Version'ı güncelle
        node.Version = CreateVersion()
    }
    
    r.nodes[node.ID] = node
    r.health[node.ID] = time.Now()
    
    // Aboneleri bilgilendir
    r.notifySubscribers(ResourceUpdate{
        Action:   action,
        Type:     "node",
        Resource: node,
    })
    
    return nil
}

// RegisterDevice, RegisterSource, RegisterFlow, RegisterSender, RegisterReceiver
// Hepsi aynı pattern'i takip eder. İşte generic implementasyon:

func (r *Registry) registerResource(id string, resource interface{}, resourceType string, 
    resourceMap map[string]interface{}, validate func() error) error {
    r.mu.Lock()
    defer r.mu.Unlock()
    
    // Opsiyonel validasyon
    if validate != nil {
        if err := validate(); err != nil {
            return err
        }
    }
    
    action := "create"
    if _, exists := resourceMap[id]; exists {
        action = "update"
        // Version'ı güncelle (kaynak türüne bağlı implementasyon)
    }
    
    resourceMap[id] = resource
    
    r.notifySubscribers(ResourceUpdate{
        Action:   action,
        Type:     resourceType,
        Resource: resource,
    })
    
    return nil
}

// Kolaylık wrapper'ları:
func (r *Registry) RegisterDevice(device *Device) error {
    return r.registerResource(device.ID, device, "device", 
        cast(r.devices), func() error {
            if _, exists := r.nodes[device.NodeID]; !exists {
                return fmt.Errorf("node %s bulunamadı", device.NodeID)
            }
            return nil
        })
}

func (r *Registry) RegisterSender(sender *Sender) error {
    return r.registerResource(sender.ID, sender, "sender", cast(r.senders), nil)
}

func (r *Registry) RegisterReceiver(receiver *Receiver) error {
    return r.registerResource(receiver.ID, receiver, "receiver", cast(r.receivers), nil)
}

// RegisterSource, RegisterFlow için benzer şekilde...

// Query metotları - Generic getter
func getResources[T any](mu *sync.RWMutex, resourceMap map[string]*T) []*T {
    mu.RLock()
    defer mu.RUnlock()
    
    resources := make([]*T, 0, len(resourceMap))
    for _, resource := range resourceMap {
        resources = append(resources, resource)
    }
    return resources
}

// Kolaylık metotları
func (r *Registry) GetNodes() []*Node {
    return getResources(&r.mu, r.nodes)
}

func (r *Registry) GetSenders() []*Sender {
    return getResources(&r.mu, r.senders)
}

func (r *Registry) GetReceivers() []*Receiver {
    return getResources(&r.mu, r.receivers)
}

// GetDevices(), GetSources(), GetFlows() için benzer şekilde...

// Sağlık kontrolü - Eski node'ları temizle
func (r *Registry) healthCheckLoop() {
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()
    
    for range ticker.C {
        r.checkHealth()
    }
}

func (r *Registry) checkHealth() {
    r.mu.Lock()
    defer r.mu.Unlock()
    
    staleThreshold := 12 * time.Second // IS-04 spec: node'lar her 5s'de heartbeat göndermeli
    now := time.Now()
    
    for nodeID, lastSeen := range r.health {
        if now.Sub(lastSeen) > staleThreshold {
            // Eski node ve kaynaklarını kaldır
            delete(r.nodes, nodeID)
            delete(r.health, nodeID)
            
            // İlişkili device, sender, receiver'ları kaldır
            r.removeNodeResources(nodeID)
            
            fmt.Printf("Eski node kaldırıldı: %s\n", nodeID)
        }
    }
}

func (r *Registry) removeNodeResources(nodeID string) {
    // Device'ları kaldır
    for id, device := range r.devices {
        if device.NodeID == nodeID {
            delete(r.devices, id)
        }
    }
    
    // Kaldırılan device'lara ait sender/receiver'ları kaldır
    // (basitleştirilmiş - production'da device ilişkilerini takip edin)
}

// Heartbeat, node sağlığını günceller
func (r *Registry) Heartbeat(nodeID string) error {
    r.mu.Lock()
    defer r.mu.Unlock()
    
    if _, exists := r.nodes[nodeID]; !exists {
        return fmt.Errorf("node %s kayıtlı değil", nodeID)
    }
    
    r.health[nodeID] = time.Now()
    return nil
}

// Subscriber yönetimi
func (r *Registry) Subscribe(query SubscriptionQuery) *Subscriber {
    r.mu.Lock()
    defer r.mu.Unlock()
    
    sub := &Subscriber{
        ID:      generateUUID(),
        Query:   query,
        Updates: make(chan ResourceUpdate, 100),
    }
    
    r.subscribers[sub.ID] = sub
    return sub
}

func (r *Registry) Unsubscribe(subID string) {
    r.mu.Lock()
    defer r.mu.Unlock()
    
    if sub, exists := r.subscribers[subID]; exists {
        close(sub.Updates)
        delete(r.subscribers, subID)
    }
}

func (r *Registry) notifySubscribers(update ResourceUpdate) {
    for _, sub := range r.subscribers {
        // Subscription query'sine göre filtrele
        if r.matchesQuery(update, sub.Query) {
            select {
            case sub.Updates <- update:
            default:
                // Subscriber yavaş, atla
            }
        }
    }
}

func (r *Registry) matchesQuery(update ResourceUpdate, query SubscriptionQuery) bool {
    // Kaynak türünü eşleştir
    if query.ResourceType != "" && query.ResourceType != update.Type {
        return false
    }
    
    // Ek sorgu parametrelerini eşleştir
    // (basitleştirilmiş - production'da tam sorgu eşleştirmesi implement edin)
    
    return true
}

func generateUUID() string {
    // Production'da github.com/google/uuid kullanın
    return fmt.Sprintf("%d", time.Now().UnixNano())
}

  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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174

// registry/registration.go
package registry

import (
    "encoding/json"
    "fmt"
    "net/http"
    "strings"
    
    "github.com/gorilla/mux"
)

// RegistrationAPI, IS-04 Registration API'sini yönetir
type RegistrationAPI struct {
    registry *Registry
}

func NewRegistrationAPI(registry *Registry) *RegistrationAPI {
    return &RegistrationAPI{registry: registry}
}

// SetupRoutes, Registration API için HTTP route'larını yapılandırır
func (api *RegistrationAPI) SetupRoutes(router *mux.Router) {
    // IS-04 v1.3 Registration API route'ları
    v13 := router.PathPrefix("/x-nmos/registration/v1.3").Subrouter()
    
    // Kaynak kaydı
    v13.HandleFunc("/resource", api.RegisterResource).Methods("POST")
    v13.HandleFunc("/resource", api.DeleteAllResources).Methods("DELETE")
    
    // Sağlık kontrolü
    v13.HandleFunc("/health/nodes/{nodeId}", api.Heartbeat).Methods("POST")
    
    // Kaynak silme
    v13.HandleFunc("/resource/{resourceType}/{resourceId}", api.DeleteResource).Methods("DELETE")
}

// RegisterResource, kaynak kaydını işler (POST /resource)
func (api *RegistrationAPI) RegisterResource(w http.ResponseWriter, r *http.Request) {
    var rawResource map[string]interface{}
    
    if err := json.NewDecoder(r.Body).Decode(&rawResource); err != nil {
        http.Error(w, "Geçersiz JSON", http.StatusBadRequest)
        return
    }
    
    // Veri'den kaynak türünü belirle
    resourceType := inferResourceType(rawResource)
    
    var err error
    switch resourceType {
    case "node":
        var node Node
        if err = remarshal(rawResource, &node); err == nil {
            err = api.registry.RegisterNode(&node)
        }
    case "device":
        var device Device
        if err = remarshal(rawResource, &device); err == nil {
            err = api.registry.RegisterDevice(&device)
        }
    case "source":
        var source Source
        if err = remarshal(rawResource, &source); err == nil {
            err = api.registry.RegisterSource(&source)
        }
    case "flow":
        var flow Flow
        if err = remarshal(rawResource, &flow); err == nil {
            err = api.registry.RegisterFlow(&flow)
        }
    case "sender":
        var sender Sender
        if err = remarshal(rawResource, &sender); err == nil {
            err = api.registry.RegisterSender(&sender)
        }
    case "receiver":
        var receiver Receiver
        if err = remarshal(rawResource, &receiver); err == nil {
            err = api.registry.RegisterReceiver(&receiver)
        }
    default:
        http.Error(w, "Bilinmeyen kaynak türü", http.StatusBadRequest)
        return
    }
    
    if err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }
    
    w.WriteHeader(http.StatusCreated)
    json.NewEncoder(w).Encode(map[string]string{
        "status": "registered",
        "type":   resourceType,
    })
}

// Heartbeat, node sağlık kontrollerini işler
func (api *RegistrationAPI) Heartbeat(w http.ResponseWriter, r *http.Request) {
    vars := mux.Vars(r)
    nodeID := vars["nodeId"]
    
    if err := api.registry.Heartbeat(nodeID); err != nil {
        http.Error(w, err.Error(), http.StatusNotFound)
        return
    }
    
    w.WriteHeader(http.StatusOK)
    json.NewEncoder(w).Encode(map[string]string{
        "status": "healthy",
    })
}

// DeleteResource, kaynak silme işlemini yönetir
func (api *RegistrationAPI) DeleteResource(w http.ResponseWriter, r *http.Request) {
    vars := mux.Vars(r)
    resourceType := vars["resourceType"]
    resourceID := vars["resourceId"]
    
    if err := api.registry.DeleteResource(resourceType, resourceID); err != nil {
        http.Error(w, err.Error(), http.StatusNotFound)
        return
    }
    
    w.WriteHeader(http.StatusNoContent)
}

// DeleteAllResources, bir node için tüm kaynakları kaldırır
func (api *RegistrationAPI) DeleteAllResources(w http.ResponseWriter, r *http.Request) {
    // Bu, bir node graceful shutdown yaparken çağrılır
    w.WriteHeader(http.StatusNoContent)
}

// Yardımcı fonksiyonlar
func inferResourceType(data map[string]interface{}) string {
    // NMOS kaynakları farklı zorunlu alanlara sahiptir
    if _, hasHostname := data["hostname"]; hasHostname {
        return "node"
    }
    if _, hasManifest := data["manifest_href"]; hasManifest {
        return "sender"
    }
    if caps, hasCaps := data["caps"]; hasCaps {
        if capsMap, ok := caps.(map[string]interface{}); ok {
            if _, hasMediaTypes := capsMap["media_types"]; hasMediaTypes {
                return "receiver"
            }
        }
        return "source"
    }
    if _, hasTransport := data["transport"]; hasTransport {
        if _, hasFlowID := data["flow_id"]; hasFlowID {
            return "sender"
        }
        return "receiver"
    }
    if _, hasSourceID := data["source_id"]; hasSourceID {
        return "flow"
    }
    if _, hasNodeID := data["node_id"]; hasNodeID {
        return "device"
    }
    
    return "unknown"
}

func remarshal(from interface{}, to interface{}) error {
    data, err := json.Marshal(from)
    if err != nil {
        return err
    }
    return json.Unmarshal(data, to)
}

 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

# 1. Node Kaydet
curl -X POST http://localhost:3210/x-nmos/registration/v1.3/resource \
  -H "Content-Type: application/json" \
  -d '{
    "id": "58f6b536-ca4c-43fd-880e-9df2af5d5d94",
    "version": "1735545600:0",
    "label": "Camera Encoder 01",
    "description": "Production camera encoder",
    "hostname": "encoder-01.local",
    "api": ["v1.3"],
    "caps": {},
    "services": [],
    "clocks": [{"name": "clk0", "ref_type": "ptp"}],
    "interfaces": [{"name": "eth0"}],
    "tags": {}
  }'

# 2. Sender Kaydet
curl -X POST http://localhost:3210/x-nmos/registration/v1.3/resource \
  -H "Content-Type: application/json" \
  -d '{
    "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "version": "1735545600:0",
    "label": "Camera 1 Video",
    "description": "1080p59.94 video sender",
    "flow_id": "f1f2f3f4-f5f6-f7f8-f9f0-f1f2f3f4f5f6",
    "transport": "urn:x-nmos:transport:rtp",
    "device_id": "d1d2d3d4-d5d6-d7d8-d9d0-d1d2d3d4d5d6",
    "manifest_href": "http://encoder-01.local:8080/sdp/camera1.sdp",
    "interface_bindings": ["eth0"],
    "tags": {}
  }'

# 3. Heartbeat Gönder
NODE_ID="58f6b536-ca4c-43fd-880e-9df2af5d5d94"
curl -X POST http://localhost:3210/x-nmos/registration/v1.3/health/nodes/$NODE_ID

# 4. Kaynak Sil
curl -X DELETE http://localhost:3210/x-nmos/registration/v1.3/resource/senders/a1b2c3d4-e5f6-7890-abcd-ef1234567890

# Response Örnekleri:
# Başarılı (201 Created):
# {"status":"registered","type":"node"}

# Heartbeat Başarılı (200 OK):
# {"status":"healthy"}

  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

// registry/query.go
package registry

import (
    "encoding/json"
    "net/http"
    "strings"
    
    "github.com/gorilla/mux"
)

// QueryAPI, IS-04 Query API'sini yönetir
type QueryAPI struct {
    registry *Registry
}

func NewQueryAPI(registry *Registry) *QueryAPI {
    return &QueryAPI{registry: registry}
}

// SetupRoutes, Query API için HTTP route'larını yapılandırır
func (api *QueryAPI) SetupRoutes(router *mux.Router) {
    // IS-04 v1.3 Query API route'ları
    v13 := router.PathPrefix("/x-nmos/query/v1.3").Subrouter()
    
    // Kaynak sorguları
    v13.HandleFunc("/nodes", api.GetNodes).Methods("GET")
    v13.HandleFunc("/devices", api.GetDevices).Methods("GET")
    v13.HandleFunc("/sources", api.GetSources).Methods("GET")
    v13.HandleFunc("/flows", api.GetFlows).Methods("GET")
    v13.HandleFunc("/senders", api.GetSenders).Methods("GET")
    v13.HandleFunc("/receivers", api.GetReceivers).Methods("GET")
    
    // Tek kaynak sorgusu
    v13.HandleFunc("/nodes/{nodeId}", api.GetNode).Methods("GET")
    v13.HandleFunc("/senders/{senderId}", api.GetSender).Methods("GET")
    v13.HandleFunc("/receivers/{receiverId}", api.GetReceiver).Methods("GET")
    
    // WebSocket subscription
    v13.HandleFunc("/subscriptions", api.CreateSubscription).Methods("POST")
    v13.HandleFunc("/subscriptions/{subscriptionId}", api.DeleteSubscription).Methods("DELETE")
}

// GetNodes, tüm kayıtlı node'ları döndürür
func (api *QueryAPI) GetNodes(w http.ResponseWriter, r *http.Request) {
    nodes := api.registry.GetNodes()
    
    // Query parametrelerini uygula (label, description filtreleri)
    nodes = api.filterNodes(nodes, r.URL.Query())
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(nodes)
}

// GetSenders, tüm kayıtlı sender'ları döndürür
func (api *QueryAPI) GetSenders(w http.ResponseWriter, r *http.Request) {
    senders := api.registry.GetSenders()
    
    // Filtreleri uygula
    if transport := r.URL.Query().Get("transport"); transport != "" {
        filtered := []*Sender{}
        for _, s := range senders {
            if strings.Contains(s.Transport, transport) {
                filtered = append(filtered, s)
            }
        }
        senders = filtered
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(senders)
}

// GetReceivers, tüm kayıtlı receiver'ları döndürür
func (api *QueryAPI) GetReceivers(w http.ResponseWriter, r *http.Request) {
    receivers := api.registry.GetReceivers()
    
    // Filtreleri uygula
    if format := r.URL.Query().Get("format"); format != "" {
        filtered := []*Receiver{}
        for _, rcv := range receivers {
            if strings.Contains(rcv.Format, format) {
                filtered = append(filtered, rcv)
            }
        }
        receivers = filtered
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(receivers)
}

// GetNode, ID'ye göre belirli bir node döndürür
func (api *QueryAPI) GetNode(w http.ResponseWriter, r *http.Request) {
    vars := mux.Vars(r)
    nodeID := vars["nodeId"]
    
    node := api.registry.GetNode(nodeID)
    if node == nil {
        http.Error(w, "Node bulunamadı", http.StatusNotFound)
        return
    }
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(node)
}

// CreateSubscription, WebSocket subscription oluşturur
func (api *QueryAPI) CreateSubscription(w http.ResponseWriter, r *http.Request) {
    var query SubscriptionQuery
    
    if err := json.NewDecoder(r.Body).Decode(&query); err != nil {
        http.Error(w, "Geçersiz subscription sorgusu", http.StatusBadRequest)
        return
    }
    
    sub := api.registry.Subscribe(query)
    
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(http.StatusCreated)
    json.NewEncoder(w).Encode(map[string]string{
        "id":           sub.ID,
        "ws_href":      fmt.Sprintf("ws://localhost:3211/x-nmos/query/v1.3/subscriptions/%s/ws", sub.ID),
        "resource_path": fmt.Sprintf("/%s", query.ResourceType),
    })
}

// DeleteSubscription, bir subscription'ı kaldırır
func (api *QueryAPI) DeleteSubscription(w http.ResponseWriter, r *http.Request) {
    vars := mux.Vars(r)
    subID := vars["subscriptionId"]
    
    api.registry.Unsubscribe(subID)
    w.WriteHeader(http.StatusNoContent)
}

// Filtre yardımcıları
func (api *QueryAPI) filterNodes(nodes []*Node, params url.Values) []*Node {
    if label := params.Get("label"); label != "" {
        filtered := []*Node{}
        for _, n := range nodes {
            if strings.Contains(strings.ToLower(n.Label), strings.ToLower(label)) {
                filtered = append(filtered, n)
            }
        }
        return filtered
    }
    return nodes
}

  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
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193

// registry/websocket.go
package registry

import (
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "time"
    
    "github.com/gorilla/mux"
    "github.com/gorilla/websocket"
)

var upgrader = websocket.Upgrader{
    ReadBufferSize:  1024,
    WriteBufferSize: 1024,
    CheckOrigin: func(r *http.Request) bool {
        return true // Production'da origin'i doğrulayın
    },
}

// WebSocketHandler, subscription'lar için WebSocket bağlantılarını yönetir
func (api *QueryAPI) WebSocketHandler(w http.ResponseWriter, r *http.Request) {
    vars := mux.Vars(r)
    subID := vars["subscriptionId"]
    
    // Subscription'ı al
    sub := api.registry.GetSubscription(subID)
    if sub == nil {
        http.Error(w, "Subscription bulunamadı", http.StatusNotFound)
        return
    }
    
    // WebSocket'e yükselt
    conn, err := upgrader.Upgrade(w, r, nil)
    if err != nil {
        log.Printf("WebSocket upgrade başarısız: %v", err)
        return
    }
    defer conn.Close()
    
    log.Printf("WebSocket client bağlandı: subscription %s", subID)
    
    // İlk grain'i gönder (mevcut durum)
    if err := api.sendInitialGrain(conn, sub); err != nil {
        log.Printf("İlk grain gönderilemedi: %v", err)
        return
    }
    
    // Gerçek zamanlı güncellemeleri yönet
    done := make(chan struct{})
    
    // Read pump (ping/pong ve client disconnect'i yönet)
    go func() {
        defer close(done)
        conn.SetReadDeadline(time.Now().Add(60 * time.Second))
        conn.SetPongHandler(func(string) error {
            conn.SetReadDeadline(time.Now().Add(60 * time.Second))
            return nil
        })
        
        for {
            if _, _, err := conn.ReadMessage(); err != nil {
                log.Printf("WebSocket okuma hatası: %v", err)
                return
            }
        }
    }()
    
    // Write pump (güncellemeleri client'a gönder)
    ticker := time.NewTicker(30 * time.Second) // Ping aralığı
    defer ticker.Stop()
    
    for {
        select {
        case update := <-sub.Updates:
            // Kaynak güncellemeyi client'a gönder
            if err := api.sendUpdate(conn, update); err != nil {
                log.Printf("Güncelleme gönderilemedi: %v", err)
                return
            }
            
        case <-ticker.C:
            // Ping gönder
            if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
                return
            }
            
        case <-done:
            return
        }
    }
}

// sendInitialGrain, kaynakların mevcut durumunu gönderir
func (api *QueryAPI) sendInitialGrain(conn *websocket.Conn, sub *Subscriber) error {
    var resources []interface{}
    
    // Subscription sorguya göre kaynakları al
    switch sub.Query.ResourceType {
    case "nodes":
        for _, node := range api.registry.GetNodes() {
            resources = append(resources, node)
        }
    case "senders":
        for _, sender := range api.registry.GetSenders() {
            resources = append(resources, sender)
        }
    case "receivers":
        for _, receiver := range api.registry.GetReceivers() {
            resources = append(resources, receiver)
        }
    default:
        // Tüm kaynakları al
        for _, node := range api.registry.GetNodes() {
            resources = append(resources, node)
        }
    }
    
    // Grain mesajı oluştur
    grain := map[string]interface{}{
        "grain_type": "event",
        "source_id":  "nmos-registry",
        "flow_id":    "query-api-events",
        "origin_timestamp": time.Now().Format(time.RFC3339Nano),
        "sync_timestamp":   time.Now().Format(time.RFC3339Nano),
        "creation_timestamp": time.Now().Format(time.RFC3339Nano),
        "rate": map[string]int{
            "numerator":   0,
            "denominator": 1,
        },
        "duration": map[string]int{
            "numerator":   0,
            "denominator": 1,
        },
        "grain": map[string]interface{}{
            "type": "urn:x-nmos:format:data.event",
            "topic": fmt.Sprintf("/%s/", sub.Query.ResourceType),
            "data": resources,
        },
    }
    
    return conn.WriteJSON(grain)
}

// sendUpdate, kaynak değişikliği bildirimi gönderir
func (api *QueryAPI) sendUpdate(conn *websocket.Conn, update ResourceUpdate) error {
    // Event grain oluştur
    var eventType string
    switch update.Action {
    case "create":
        eventType = "resource_added"
    case "update":
        eventType = "resource_modified"
    case "delete":
        eventType = "resource_removed"
    }
    
    grain := map[string]interface{}{
        "grain_type": "event",
        "source_id":  "nmos-registry",
        "flow_id":    "query-api-events",
        "origin_timestamp": time.Now().Format(time.RFC3339Nano),
        "sync_timestamp":   time.Now().Format(time.RFC3339Nano),
        "creation_timestamp": time.Now().Format(time.RFC3339Nano),
        "rate": map[string]int{
            "numerator":   0,
            "denominator": 1,
        },
        "duration": map[string]int{
            "numerator":   0,
            "denominator": 1,
        },
        "grain": map[string]interface{}{
            "type": "urn:x-nmos:format:data.event",
            "topic": fmt.Sprintf("/%s/", update.Type),
            "data": []map[string]interface{}{
                {
                    "path": fmt.Sprintf("/%s", update.Type),
                    "pre":  nil, // Önceki durum (update/delete ise)
                    "post": update.Resource, // Yeni durum (create/update ise)
                },
            },
        },
    }
    
    return conn.WriteJSON(grain)
}

// QueryAPI.SetupRoutes'a WebSocket endpoint'i ekleyin
// Bu route'u ekleyin:
// v13.HandleFunc("/subscriptions/{subscriptionId}/ws", api.WebSocketHandler).Methods("GET")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# wscat yükle
npm install -g wscat

# Önce subscription oluştur
SUB_ID=$(curl -X POST http://localhost:3211/x-nmos/query/v1.3/subscriptions \
  -H "Content-Type: application/json" \
  -d '{"resource_type":"senders"}' | jq -r '.id')

# WebSocket'e bağlan
wscat -c "ws://localhost:3211/x-nmos/query/v1.3/subscriptions/$SUB_ID/ws"

# Sender eklendiğinde/kaldırıldığında/değiştirildiğinde gerçek zamanlı güncellemeler alacaksınız

 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

# 1. Tüm node'ları al
curl http://localhost:3211/x-nmos/query/v1.3/nodes | jq

# 2. Tüm sender'ları al
curl http://localhost:3211/x-nmos/query/v1.3/senders | jq

# 3. Belirli sender'ı al
SENDER_ID="a1b2c3d4-e5f6-7890-abcd-ef1234567890"
curl http://localhost:3211/x-nmos/query/v1.3/senders/$SENDER_ID | jq

# 4. Filtrelerle sorgula
# Sadece video sender'ları al
curl "http://localhost:3211/x-nmos/query/v1.3/senders?format=urn:x-nmos:format:video" | jq

# Label'a göre node'ları al
curl "http://localhost:3211/x-nmos/query/v1.3/nodes?label=camera" | jq

# 5. WebSocket subscription oluştur
curl -X POST http://localhost:3211/x-nmos/query/v1.3/subscriptions \
  -H "Content-Type: application/json" \
  -d '{
    "resource_type": "senders",
    "params": {}
  }' | jq

# Response:
# {
#   "id": "sub-12345",
#   "ws_href": "ws://localhost:3211/x-nmos/query/v1.3/subscriptions/sub-12345/ws",
#   "resource_path": "/senders"
# }

# 6. Pagination (büyük registry'ler için)
curl "http://localhost:3211/x-nmos/query/v1.3/senders?paging.limit=10&paging.offset=0" | jq

 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

// main.go
package main

import (
    "fmt"
    "log"
    "net/http"
    
    "github.com/gorilla/mux"
    "yourproject/registry"
)

func main() {
    // Registry oluştur
    reg := registry.NewRegistry()
    
    // API'leri kur
    registrationAPI := registry.NewRegistrationAPI(reg)
    queryAPI := registry.NewQueryAPI(reg)
    
    // Router oluştur
    router := mux.NewRouter()
    
    // Route'ları kur
    registrationAPI.SetupRoutes(router)
    queryAPI.SetupRoutes(router)
    
    // Health check endpoint
    router.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
        w.WriteHeader(http.StatusOK)
        fmt.Fprintf(w, "OK")
    }).Methods("GET")
    
    // Registration API'yi port 3210'da başlat
    go func() {
        regRouter := mux.NewRouter()
        registrationAPI.SetupRoutes(regRouter)
        log.Println("Registration API :3210'da dinliyor")
        log.Fatal(http.ListenAndServe(":3210", regRouter))
    }()
    
    // Query API'yi port 3211'de başlat
    log.Println("Query API :3211'de dinliyor")
    log.Fatal(http.ListenAndServe(":3211", router))
}