281 lines
8.0 KiB
HTML
281 lines
8.0 KiB
HTML
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<meta http-equiv="Content-Type" content="text/html; charset=utf-8">
|
|
<title>storebus: Go Coverage Report</title>
|
|
<style>
|
|
body {
|
|
background: black;
|
|
color: rgb(80, 80, 80);
|
|
}
|
|
body, pre, #legend span {
|
|
font-family: Menlo, monospace;
|
|
font-weight: bold;
|
|
}
|
|
#topbar {
|
|
background: black;
|
|
position: fixed;
|
|
top: 0; left: 0; right: 0;
|
|
height: 42px;
|
|
border-bottom: 1px solid rgb(80, 80, 80);
|
|
}
|
|
#content {
|
|
margin-top: 50px;
|
|
}
|
|
#nav, #legend {
|
|
float: left;
|
|
margin-left: 10px;
|
|
}
|
|
#legend {
|
|
margin-top: 12px;
|
|
}
|
|
#nav {
|
|
margin-top: 10px;
|
|
}
|
|
#legend span {
|
|
margin: 0 5px;
|
|
}
|
|
.cov0 { color: rgb(192, 0, 0) }
|
|
.cov1 { color: rgb(128, 128, 128) }
|
|
.cov2 { color: rgb(116, 140, 131) }
|
|
.cov3 { color: rgb(104, 152, 134) }
|
|
.cov4 { color: rgb(92, 164, 137) }
|
|
.cov5 { color: rgb(80, 176, 140) }
|
|
.cov6 { color: rgb(68, 188, 143) }
|
|
.cov7 { color: rgb(56, 200, 146) }
|
|
.cov8 { color: rgb(44, 212, 149) }
|
|
.cov9 { color: rgb(32, 224, 152) }
|
|
.cov10 { color: rgb(20, 236, 155) }
|
|
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div id="topbar">
|
|
<div id="nav">
|
|
<select id="files">
|
|
|
|
<option value="file0">github.com/gopatchy/storebus/storebus.go (85.5%)</option>
|
|
|
|
</select>
|
|
</div>
|
|
<div id="legend">
|
|
<span>not tracked</span>
|
|
|
|
<span class="cov0">no coverage</span>
|
|
<span class="cov1">low coverage</span>
|
|
<span class="cov2">*</span>
|
|
<span class="cov3">*</span>
|
|
<span class="cov4">*</span>
|
|
<span class="cov5">*</span>
|
|
<span class="cov6">*</span>
|
|
<span class="cov7">*</span>
|
|
<span class="cov8">*</span>
|
|
<span class="cov9">*</span>
|
|
<span class="cov10">high coverage</span>
|
|
|
|
</div>
|
|
</div>
|
|
<div id="content">
|
|
|
|
<pre class="file" id="file0" style="display: none">package storebus
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/json"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"github.com/gopatchy/bus"
|
|
"github.com/gopatchy/jsrest"
|
|
"github.com/gopatchy/metadata"
|
|
"github.com/gopatchy/store"
|
|
)
|
|
|
|
type StoreBus struct {
|
|
store *store.Store
|
|
bus *bus.Bus
|
|
|
|
// This lock ensures that no writes interleave with read/subscribe pairs
|
|
orderMu sync.RWMutex
|
|
|
|
chanMap map[<-chan []any]<-chan any
|
|
chanMapMu sync.Mutex
|
|
}
|
|
|
|
func NewStoreBus(dbname string) (*StoreBus, error) <span class="cov5" title="2">{
|
|
st, err := store.NewStore(dbname)
|
|
if err != nil </span><span class="cov0" title="0">{
|
|
return nil, err
|
|
}</span>
|
|
|
|
<span class="cov5" title="2">return &StoreBus{
|
|
store: st,
|
|
bus: bus.NewBus(),
|
|
chanMap: map[<-chan []any]<-chan any{},
|
|
}, nil</span>
|
|
}
|
|
|
|
func (sb *StoreBus) Close() <span class="cov5" title="2">{
|
|
sb.store.Close()
|
|
}</span>
|
|
|
|
func (sb *StoreBus) Write(ctx context.Context, t string, obj any) error <span class="cov10" title="4">{
|
|
sb.orderMu.Lock()
|
|
defer sb.orderMu.Unlock()
|
|
|
|
if err := UpdateHash(obj); err != nil </span><span class="cov0" title="0">{
|
|
return jsrest.Errorf(jsrest.ErrInternalServerError, "hash update failed (%w)", err)
|
|
}</span>
|
|
|
|
<span class="cov10" title="4">if err := sb.store.Write(ctx, t, obj); err != nil </span><span class="cov0" title="0">{
|
|
return jsrest.Errorf(jsrest.ErrInternalServerError, "write failed (%w)", err)
|
|
}</span>
|
|
|
|
<span class="cov10" title="4">sb.bus.Announce(t, obj)
|
|
|
|
return nil</span>
|
|
}
|
|
|
|
func (sb *StoreBus) Delete(ctx context.Context, t, id string) error <span class="cov1" title="1">{
|
|
sb.orderMu.Lock()
|
|
defer sb.orderMu.Unlock()
|
|
|
|
if err := sb.store.Delete(ctx, t, id); err != nil </span><span class="cov0" title="0">{
|
|
return jsrest.Errorf(jsrest.ErrInternalServerError, "delete failed (%w)", err)
|
|
}</span>
|
|
|
|
<span class="cov1" title="1">sb.bus.Delete(t, id)
|
|
|
|
return nil</span>
|
|
}
|
|
|
|
func (sb *StoreBus) Read(ctx context.Context, t, id string, factory func() any) (any, error) <span class="cov1" title="1">{
|
|
return sb.store.Read(ctx, t, id, factory)
|
|
}</span>
|
|
|
|
func (sb *StoreBus) ReadStream(ctx context.Context, t, id string, factory func() any) (<-chan any, error) <span class="cov5" title="2">{
|
|
sb.orderMu.RLock()
|
|
defer sb.orderMu.RUnlock()
|
|
|
|
initial, err := sb.store.Read(ctx, t, id, factory)
|
|
if err != nil </span><span class="cov0" title="0">{
|
|
return nil, jsrest.Errorf(jsrest.ErrInternalServerError, "read failed (%w)", err)
|
|
}</span>
|
|
|
|
<span class="cov5" title="2">c := sb.bus.SubscribeKey(t, id, initial)
|
|
|
|
return c, nil</span>
|
|
}
|
|
|
|
func (sb *StoreBus) CloseReadStream(t, id string, c <-chan any) <span class="cov5" title="2">{
|
|
sb.bus.UnsubscribeKey(t, id, c)
|
|
}</span>
|
|
|
|
func (sb *StoreBus) List(ctx context.Context, t string, factory func() any) ([]any, error) <span class="cov5" title="2">{
|
|
return sb.store.List(ctx, t, factory)
|
|
}</span>
|
|
|
|
func (sb *StoreBus) ListStream(ctx context.Context, t string, factory func() any) (<-chan []any, error) <span class="cov1" title="1">{
|
|
sb.orderMu.RLock()
|
|
defer sb.orderMu.RUnlock()
|
|
|
|
initial, err := sb.store.List(ctx, t, factory)
|
|
if err != nil </span><span class="cov0" title="0">{
|
|
return nil, jsrest.Errorf(jsrest.ErrInternalServerError, "list failed (%w)", err)
|
|
}</span>
|
|
|
|
<span class="cov1" title="1">c := sb.bus.SubscribeType(t, initial)
|
|
|
|
ret := make(chan []any, 100)
|
|
|
|
sb.registerChan(c, ret)
|
|
|
|
go func() </span><span class="cov1" title="1">{
|
|
defer close(ret)
|
|
|
|
for range c </span><span class="cov8" title="3">{
|
|
// List() results are always at least (but not exactly) as new as the write that triggered it
|
|
l, err := sb.store.List(ctx, t, factory)
|
|
if err != nil </span><span class="cov0" title="0">{
|
|
break</span>
|
|
}
|
|
|
|
<span class="cov8" title="3">select </span>{
|
|
case ret <- l:<span class="cov8" title="3"></span>
|
|
default:<span class="cov0" title="0">
|
|
break</span>
|
|
}
|
|
}
|
|
}()
|
|
|
|
<span class="cov1" title="1">return ret, nil</span>
|
|
}
|
|
|
|
func (sb *StoreBus) CloseListStream(t string, c <-chan []any) <span class="cov1" title="1">{
|
|
sb.chanMapMu.Lock()
|
|
defer sb.chanMapMu.Unlock()
|
|
|
|
sb.bus.UnsubscribeType(t, sb.chanMap[c])
|
|
|
|
delete(sb.chanMap, c)
|
|
}</span>
|
|
|
|
func (sb *StoreBus) registerChan(in <-chan any, out <-chan []any) <span class="cov1" title="1">{
|
|
sb.chanMapMu.Lock()
|
|
defer sb.chanMapMu.Unlock()
|
|
|
|
sb.chanMap[out] = in
|
|
}</span>
|
|
|
|
func UpdateHash(obj any) error <span class="cov10" title="4">{
|
|
m := *metadata.GetMetadata(obj)
|
|
metadata.ClearMetadata(obj)
|
|
|
|
defer metadata.SetMetadata(obj, &m)
|
|
|
|
hash := sha256.New()
|
|
enc := json.NewEncoder(hash)
|
|
|
|
if err := enc.Encode(obj); err != nil </span><span class="cov0" title="0">{
|
|
return jsrest.Errorf(jsrest.ErrInternalServerError, "JSON encode failed (%w)", err)
|
|
}</span>
|
|
|
|
<span class="cov10" title="4">m.ETag = fmt.Sprintf("etag:%x", hash.Sum(nil))
|
|
|
|
return nil</span>
|
|
}
|
|
</pre>
|
|
|
|
</div>
|
|
</body>
|
|
<script>
|
|
(function() {
|
|
var files = document.getElementById('files');
|
|
var visible;
|
|
files.addEventListener('change', onChange, false);
|
|
function select(part) {
|
|
if (visible)
|
|
visible.style.display = 'none';
|
|
visible = document.getElementById(part);
|
|
if (!visible)
|
|
return;
|
|
files.value = part;
|
|
visible.style.display = 'block';
|
|
location.hash = part;
|
|
}
|
|
function onChange() {
|
|
select(files.value);
|
|
window.scrollTo(0, 0);
|
|
}
|
|
if (location.hash != "") {
|
|
select(location.hash.substr(1));
|
|
}
|
|
if (!visible) {
|
|
select("file0");
|
|
}
|
|
})();
|
|
</script>
|
|
</html>
|