fix subscription grouping to not include filter keys
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
import type {Event} from 'nostr-tools'
|
||||
import {Emitter, randomId, groupBy, batch, defer, uniq, uniqBy} from '@coracle.social/lib'
|
||||
import {Emitter, now, randomId, groupBy, batch, defer, uniq, uniqBy} from '@coracle.social/lib'
|
||||
import type {Deferred} from '@coracle.social/lib'
|
||||
import {asEvent,} from '@coracle.social/util'
|
||||
import {Tracker} from "./Tracker"
|
||||
@@ -24,6 +24,7 @@ export type PublishRequest = {
|
||||
|
||||
export type Publish = {
|
||||
id: string
|
||||
created_at: number
|
||||
emitter: Emitter
|
||||
request: PublishRequest
|
||||
status: PublishStatusMap
|
||||
@@ -32,11 +33,12 @@ export type Publish = {
|
||||
|
||||
export const makePublish = (request: PublishRequest) => {
|
||||
const id = randomId()
|
||||
const created_at = now()
|
||||
const emitter = new Emitter()
|
||||
const result: Publish['result'] = defer()
|
||||
const status: Publish['status'] = new Map()
|
||||
|
||||
return {id, request, emitter, result, status}
|
||||
return {id, created_at, request, emitter, result, status}
|
||||
}
|
||||
|
||||
export const publish = (request: PublishRequest) => {
|
||||
@@ -58,9 +60,11 @@ export const publish = (request: PublishRequest) => {
|
||||
})
|
||||
|
||||
// Start everything off as pending
|
||||
for (const relay of request.relays) {
|
||||
pub.emitter.emit(PublishStatus.Pending, relay)
|
||||
}
|
||||
requestAnimationFrame(() => {
|
||||
for (const relay of request.relays) {
|
||||
pub.emitter.emit(PublishStatus.Pending, relay)
|
||||
}
|
||||
})
|
||||
|
||||
// Set a timeout
|
||||
const timeout = setTimeout(() => {
|
||||
|
||||
@@ -63,7 +63,7 @@ export const makeSubscription = (request: SubscribeRequest) => {
|
||||
}
|
||||
|
||||
export const calculateSubscriptionGroup = (sub: Subscription) => {
|
||||
const parts: string[] = sub.request.filters.map(calculateFilterGroup)
|
||||
const parts: string[] = []
|
||||
|
||||
if (sub.request.timeout) parts.push(`timeout:${sub.request.timeout}`)
|
||||
if (sub.request.closeOnEose) parts.push('closeOnEose')
|
||||
@@ -166,11 +166,11 @@ export const mergeSubscriptions = (subs: Subscription[]) => {
|
||||
}
|
||||
}
|
||||
|
||||
console.log(
|
||||
`Starting ${mergedSubscriptions.length} subscriptions on ${uniq(mergedSubscriptions.flatMap(s => s.request.relays)).length} relays`,
|
||||
uniq(mergedSubscriptions.flatMap(s => s.request.relays)),
|
||||
...mergeFilters(mergedSubscriptions.flatMap(s => s.request.filters)),
|
||||
)
|
||||
// console.log(
|
||||
// `Starting ${mergedSubscriptions.length} subscriptions on ${uniq(mergedSubscriptions.flatMap(s => s.request.relays)).length} relays`,
|
||||
// uniq(mergedSubscriptions.flatMap(s => s.request.relays)),
|
||||
// ...mergeFilters(mergedSubscriptions.flatMap(s => s.request.filters)),
|
||||
// )
|
||||
|
||||
return mergedSubscriptions
|
||||
}
|
||||
@@ -255,7 +255,7 @@ export const executeSubscription = (sub: Subscription) => {
|
||||
export const executeSubscriptions = (subs: Subscription[]) =>
|
||||
mergeSubscriptions(subs).forEach(executeSubscription)
|
||||
|
||||
export const executeSubscriptionBatched = batch(300, executeSubscriptions)
|
||||
export const executeSubscriptionBatched = batch(800, executeSubscriptions)
|
||||
|
||||
export const subscribe = (request: SubscribeRequest) => {
|
||||
const subscription: Subscription = makeSubscription(request)
|
||||
@@ -265,11 +265,11 @@ export const subscribe = (request: SubscribeRequest) => {
|
||||
}
|
||||
|
||||
if (request.immediate) {
|
||||
console.log(
|
||||
`Starting 1 subscriptions on ${request.relays.length} relays`,
|
||||
request.relays,
|
||||
...mergeFilters(request.filters)
|
||||
)
|
||||
// console.log(
|
||||
// `Starting 1 subscriptions on ${request.relays.length} relays`,
|
||||
// request.relays,
|
||||
// ...mergeFilters(request.filters)
|
||||
// )
|
||||
|
||||
executeSubscription(subscription)
|
||||
} else {
|
||||
|
||||
Reference in New Issue
Block a user