16348821200

Committed 17 Jul 2025 03:06PM UTC coverage: 62.868% (+0.3%) from 62.54%

Build # 16348821200

Build Type

push

github

Committed by

web-flow

Commit Message

Unwrap `swarm.Msg[swarm.Event[M, T]]` for usability (#131)

* unwrap channels to deal with event
* modify core structure to carry broker internal metadata
* adapt `emit` and `listen` api for kernel event primitives
* simplify handling of swarm.Event in the kernel
* change codec to deal with swarm.Bag instead of []byte, gives better control for apps on dynamic routing
* remove broker specific codec from kernel

Run Details

185 of 268 new or added lines in 11 files covered. (69.03%)

3 existing lines in 2 files now uncovered.

1473 of 2343 relevant lines covered (62.87%)

0.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/kernel/encoding/codec.go

//
// Copyright (C) 2021 - 2024 Dmitry Kolesnikov
//
// This file may be modified and distributed under the terms
// of the Apache License Version 2.0. See the LICENSE file for details.
// https://github.com/fogfish/swarm
//

package encoding

import (
        "encoding/json"
        "time"

        "github.com/fogfish/curie/v2"
        "github.com/fogfish/golem/optics"
        "github.com/fogfish/guid/v2"
        "github.com/fogfish/swarm"
)

//------------------------------------------------------------------------------

// JSON encoding for typed objects
type Typed[T any] string

func (c Typed[T]) Category() string { return string(c) }

func (c Typed[T]) Encode(obj T) (swarm.Bag, error) {
        msg, err := json.Marshal(obj)
        if err != nil {
                return swarm.Bag{}, err
        }

        return swarm.Bag{
                Category: string(c),
                Object:   msg,
        }, nil
}

func (Typed[T]) Decode(bag swarm.Bag) (x T, err error) {
        err = json.Unmarshal(bag.Object, &x)
        return
}

// Create JSON codec for typed objects
func ForTyped[T any](category ...string) Typed[T] {
        return Typed[T](swarm.TypeOf[T](category...))
}

//------------------------------------------------------------------------------

// JSON encoding for events
type Event[M, T any] struct {
        realm curie.IRI
        agent curie.IRI
        cat   curie.IRI
        shape optics.Lens5[M, string, curie.IRI, curie.IRI, curie.IRI, time.Time]
}

func (c Event[M, T]) Category() string { return string(c.cat) }

func (c Event[M, T]) Encode(obj swarm.Event[M, T]) (swarm.Bag, error) {
        if obj.Meta == nil {
                obj.Meta = new(M)
        }

        _, cat, rlm, agt, _ := c.shape.Get(obj.Meta)
        if cat == "" {
                cat = c.cat
        }

        if agt == "" {
                agt = c.agent
        }

        if rlm == "" {
                rlm = c.realm
        }

        c.shape.Put(obj.Meta, guid.G(guid.Clock).String(), cat, rlm, agt, time.Now())
        msg, err := json.Marshal(obj)
        if err != nil {
                return swarm.Bag{}, err
        }

        return swarm.Bag{
                Category: string(cat),
                Object:   msg,
        }, nil
}

func (c Event[M, T]) Decode(bag swarm.Bag) (swarm.Event[M, T], error) {
        var x swarm.Event[M, T]
        err := json.Unmarshal(bag.Object, &x)

        return x, err
}

// Creates JSON codec for events
func ForEvent[E swarm.Event[M, T], M, T any](realm, agent string, category ...string) Event[M, T] {
        return Event[M, T]{
                realm: curie.IRI(realm),
                agent: curie.IRI(agent),
                cat:   curie.IRI(swarm.TypeOf[T](category...)),
                shape: optics.ForShape5[M, string, curie.IRI, curie.IRI, curie.IRI, time.Time]("ID", "Type", "Realm", "Agent", "Created"),
        }
}

//------------------------------------------------------------------------------

// Identity encoding for bytes
type Bytes string

func (c Bytes) Category() string { return string(c) }

func (c Bytes) Encode(x []byte) (swarm.Bag, error) {
        return swarm.Bag{
                Category: string(c),
                Object:   x,
        }, nil
}

func (Bytes) Decode(bag swarm.Bag) ([]byte, error) {
        return bag.Object, nil
}

// Create bytes identity codec
func ForBytes(cat string) Bytes { return Bytes(cat) }

1	//
2	// Copyright (C) 2021 - 2024 Dmitry Kolesnikov
3	//
4	// This file may be modified and distributed under the terms
5	// of the Apache License Version 2.0. See the LICENSE file for details.
6	// https://github.com/fogfish/swarm
7	//
8
9	package encoding
10
11	import (
12	"encoding/json"
13	"time"
14
15	"github.com/fogfish/curie/v2"
16	"github.com/fogfish/golem/optics"
17	"github.com/fogfish/guid/v2"
18	"github.com/fogfish/swarm"
19	)
20
21	//------------------------------------------------------------------------------
22
23	// JSON encoding for typed objects
24	type Typed[T any] string
25
26	func (c Typed[T]) Category() string { return string(c) }	×
27
NEW 28	func (c Typed[T]) Encode(obj T) (swarm.Bag, error) {	×
NEW 29	msg, err := json.Marshal(obj)	×
NEW 30	if err != nil {	×
NEW 31	return swarm.Bag{}, err	×
NEW 32	}	×
33
NEW 34	return swarm.Bag{	×
NEW 35	Category: string(c),	×
NEW 36	Object: msg,	×
NEW 37	}, nil	×
38	}
39
NEW 40	func (Typed[T]) Decode(bag swarm.Bag) (x T, err error) {	×
NEW 41	err = json.Unmarshal(bag.Object, &x)	×
42	return	×
43	}	×
44
45	// Create JSON codec for typed objects
46	func ForTyped[T any](category ...string) Typed[T] {	×
47	return Typed[T](swarm.TypeOf[T](category...))	×
48	}	×
49
50	//------------------------------------------------------------------------------
51
52	// JSON encoding for events
53	type Event[M, T any] struct {
54	realm curie.IRI
55	agent curie.IRI
56	cat curie.IRI
57	shape optics.Lens5[M, string, curie.IRI, curie.IRI, curie.IRI, time.Time]
58	}
59
60	func (c Event[M, T]) Category() string { return string(c.cat) }	×
61
NEW 62	func (c Event[M, T]) Encode(obj swarm.Event[M, T]) (swarm.Bag, error) {	×
NEW 63	if obj.Meta == nil {	×
NEW 64	obj.Meta = new(M)	×
NEW 65	}	×
66
67	_, cat, rlm, agt, _ := c.shape.Get(obj.Meta)	×
68	if cat == "" {	×
69	cat = c.cat	×
70	}	×
71
72	if agt == "" {	×
73	agt = c.agent	×
74	}	×
75
76	if rlm == "" {	×
77	rlm = c.realm	×
78	}	×
79
80	c.shape.Put(obj.Meta, guid.G(guid.Clock).String(), cat, rlm, agt, time.Now())	×
NEW 81	msg, err := json.Marshal(obj)	×
NEW 82	if err != nil {	×
NEW 83	return swarm.Bag{}, err	×
NEW 84	}	×
85
NEW 86	return swarm.Bag{	×
NEW 87	Category: string(cat),	×
NEW 88	Object: msg,	×
NEW 89	}, nil	×
90	}
91
NEW 92	func (c Event[M, T]) Decode(bag swarm.Bag) (swarm.Event[M, T], error) {	×
93	var x swarm.Event[M, T]	×
NEW 94	err := json.Unmarshal(bag.Object, &x)	×
95		×
96	return x, err	×
97	}	×
98
99	// Creates JSON codec for events
100	func ForEvent[E swarm.Event[M, T], M, T any](realm, agent string, category ...string) Event[M, T] {	×
101	return Event[M, T]{	×
102	realm: curie.IRI(realm),	×
103	agent: curie.IRI(agent),	×
104	cat: curie.IRI(swarm.TypeOf[T](category...)),	×
105	shape: optics.ForShape5[M, string, curie.IRI, curie.IRI, curie.IRI, time.Time]("ID", "Type", "Realm", "Agent", "Created"),	×
106	}	×
107	}	×
108
109	//------------------------------------------------------------------------------
110
111	// Identity encoding for bytes
112	type Bytes string
113
NEW 114	func (c Bytes) Category() string { return string(c) }	×
115
NEW 116	func (c Bytes) Encode(x []byte) (swarm.Bag, error) {	×
NEW 117	return swarm.Bag{	×
NEW 118	Category: string(c),	×
NEW 119	Object: x,	×
NEW 120	}, nil	×
UNCOV 121	}	×
122
NEW 123	func (Bytes) Decode(bag swarm.Bag) ([]byte, error) {	×
NEW 124	return bag.Object, nil	×
UNCOV 125	}	×
126
127	// Create bytes identity codec
NEW 128	func ForBytes(cat string) Bytes { return Bytes(cat) }	×

fogfish / swarm / 16348821200

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous