{{ $name }}
{{ $stat.NRows }}
{{ $stat.Latency }}
@@ -160,13 +159,13 @@
E2PG
let updates = new EventSource("/task-updates");
updates.onmessage = function(event) {
const tu = JSON.parse(event.data);
- update(tu.ID, "Num", comma(tu.Num));
- update(tu.ID, "Hash", tu.Hash);
- update(tu.ID, "Latency", tu.Latency);
- update(tu.ID, "NRows", tu.NRows);
+ update(tu.SrcName, "Num", comma(tu.Num));
+ update(tu.SrcName, "Hash", tu.Hash);
+ update(tu.SrcName, "Latency", tu.Latency);
+ update(tu.SrcName, "NRows", tu.NRows);
for (const [k1, v1] of Object.entries(tu.Dstat)) {
for (const [k2, v2] of Object.entries(v1)) {
- update(`${tu.ID}-${k1}`, k2, v2);
+ update(`${tu.SrcName}-${k1}`, k2, v2);
}
}
};
diff --git a/e2pg/web/web.go b/e2pg/web/web.go
index 5b127e82..9ed5c5ef 100644
--- a/e2pg/web/web.go
+++ b/e2pg/web/web.go
@@ -40,7 +40,7 @@ type Handler struct {
conf *e2pg.Config
clientsMutex sync.Mutex
- clients map[string]chan string
+ clients map[string]chan uint64
templates map[string]*template.Template
}
@@ -50,7 +50,7 @@ func New(mgr *e2pg.Manager, conf *e2pg.Config, pgp *pgxpool.Pool) *Handler {
pgp: pgp,
mgr: mgr,
conf: conf,
- clients: make(map[string]chan string),
+ clients: make(map[string]chan uint64),
templates: make(map[string]*template.Template),
}
go func() {
@@ -167,7 +167,7 @@ func (h *Handler) Index(w http.ResponseWriter, r *http.Request) {
}
view.TaskUpdates = make(map[string]e2pg.TaskUpdate)
for _, tu := range tus {
- view.TaskUpdates[tu.ID] = tu
+ view.TaskUpdates[tu.SrcName] = tu
}
view.SourceConfigs, err = h.conf.AllSourceConfigs(ctx, h.pgp)
if err != nil {
@@ -192,7 +192,7 @@ func (h *Handler) Updates(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Connection", "keep-alive")
slog.InfoContext(r.Context(), "start sse", "c", r.RemoteAddr, "n", len(h.clients))
- c := make(chan string)
+ c := make(chan uint64)
h.clientsMutex.Lock()
h.clients[r.RemoteAddr] = c
h.clientsMutex.Unlock()
@@ -205,29 +205,30 @@ func (h *Handler) Updates(w http.ResponseWriter, r *http.Request) {
}()
for {
- var tid string
select {
- case tid = <-c:
+ case <-c:
case <-r.Context().Done(): // disconnect
return
}
- tu, err := e2pg.TaskUpdate1(r.Context(), h.pgp, tid)
+ tus, err := e2pg.TaskUpdates(r.Context(), h.pgp)
if err != nil {
slog.ErrorContext(r.Context(), "json error", "e", err)
return
}
- sjson, err := json.Marshal(tu)
- if err != nil {
- slog.ErrorContext(r.Context(), "json error", "e", err)
- return
- }
- fmt.Fprintf(w, "data: %s\n\n", sjson)
- flusher, ok := w.(http.Flusher)
- if !ok {
- http.Error(w, "Streaming unsupported", http.StatusInternalServerError)
- return
+ for i := range tus {
+ sjson, err := json.Marshal(tus[i])
+ if err != nil {
+ slog.ErrorContext(r.Context(), "json error", "e", err)
+ return
+ }
+ fmt.Fprintf(w, "data: %s\n\n", sjson)
+ flusher, ok := w.(http.Flusher)
+ if !ok {
+ http.Error(w, "Streaming unsupported", http.StatusInternalServerError)
+ return
+ }
+ flusher.Flush()
}
- flusher.Flush()
}
}
diff --git a/wctx/ctx.go b/wctx/ctx.go
index 87f1774b..b1419672 100644
--- a/wctx/ctx.go
+++ b/wctx/ctx.go
@@ -6,8 +6,9 @@ import "context"
type key int
const (
- taskIDKey key = 1
- chainIDKey key = 2
+ chainIDKey key = 1
+ intgNameKey key = 2
+ srcNameKey key = 3
)
func WithChainID(ctx context.Context, id uint64) context.Context {
@@ -18,11 +19,21 @@ func ChainID(ctx context.Context) uint64 {
id, _ := ctx.Value(chainIDKey).(uint64)
return id
}
-func WithTaskID(ctx context.Context, id string) context.Context {
- return context.WithValue(ctx, taskIDKey, id)
+
+func WithIntgName(ctx context.Context, name string) context.Context {
+ return context.WithValue(ctx, intgNameKey, name)
}
-func TaskID(ctx context.Context) string {
- id, _ := ctx.Value(taskIDKey).(string)
- return id
+func IntgName(ctx context.Context) string {
+ name, _ := ctx.Value(intgNameKey).(string)
+ return name
+}
+
+func WithSrcName(ctx context.Context, name string) context.Context {
+ return context.WithValue(ctx, srcNameKey, name)
+}
+
+func SrcName(ctx context.Context) string {
+ name, _ := ctx.Value(srcNameKey).(string)
+ return name
}