fix memory leak

src/services/mastodon.js

@@ -1,6 +1,6 @@
 import Observable from 'core-js-pure/features/observable'
 import getUrls from 'get-urls'
-import { observableToAsyncIterator, raceIterator, urlsToMedia } from '/services/misc.js'
+import { urlsToMedia } from '/services/misc.js'
 
 const LINK_RE = /<(.+?)>; rel="(\w+)"/gi
 
@@ -22,7 +22,6 @@ export async function* statusIterator({ domain, id }) {
     yield await fetchStatus(domain, id)
 }
 
-// Observable<{ domain : string, hashtag : string, status : Status}>
 export const hashtagStreamingObservable = (domain, hashtag) => {
     return new Observable(observer => {
         const onOpen = () => {
@@ -47,6 +46,7 @@ export const hashtagStreamingObservable = (domain, hashtag) => {
         eventSource.addEventListener('error', onError)
 
         return () => {
+            console.log(`Streaming ${domain} #${hashtag} : closed`)
             eventSource.removeEventListener('open', onOpen)
             eventSource.removeEventListener('update', onStatus)
             eventSource.removeEventListener('error', onError)
@@ -55,6 +55,18 @@ export const hashtagStreamingObservable = (domain, hashtag) => {
     })
 }
 
+export const hashtagsStreamingObservable = (domain, hashtags) => {
+    return new Observable(observer => {
+        const subscriptions = hashtags
+            .map(hashtag => hashtagStreamingObservable(domain, hashtag))
+            .map(observable => observable.subscribe(observer))
+
+        return () => {
+            subscriptions.forEach(subscription => subscription.unsubscribe())
+        }
+    })
+}
+
 export async function* hashtagTimelineIterator (domain, hashtag) {
     let nextLink = `https://${domain}/api/v1/timelines/tag/${hashtag}?limit=40`
 
@@ -73,32 +85,54 @@ export async function* hashtagTimelineIterator (domain, hashtag) {
     }
 }
 
-export const hashtagIterator = (domain, hashtag) => {
+export async function* hashtagsTimelineIterator (domain, hashtags) {
-    return raceIterator([
+    const iterators = hashtags.map(hashtag => hashtagTimelineIterator(domain, hashtag))
-        observableToAsyncIterator(hashtagStreamingObservable(domain, hashtag)),
+    const promises = iterators.map(iterator => iterator.next())
-        hashtagTimelineIterator(domain, hashtag)
-    ])
-}
-
-export async function* hashtagsIterator (domain, hashtags) {
-    const iterators = hashtags.map(hashtag => hashtagIterator(domain, hashtag))
-    const values = iterators.map(iterator => iterator.next())
 
     while (true) {
-        const promises = values.map((promise, index) => promise.then(result => ({ index, result })))
+        const results = (await Promise.all(promises))
-        const promisesValues = await Promise.all(promises)
+            .map((result, index) => ({ index, result }))
+            .filter(({ result }) => !result.done)
 
-        const sorted = promisesValues
+        if (results.length > 0) {
-            .sort((a, b) =>{
+            const sorted = results.sort((a, b) => b.result.value.date - a.result.value.date)
-                a.result.value.date - b.result.value.date
+            const { index, result: { value } } = sorted[0]
-            })
 
-        const { index, result: { done, value } } = sorted[0]
+            promises[index] = iterators[index].next()
-        values[index] = iterators[index].next()
+            yield value
-        yield value
+        } else {
+            break
+        }
+    }
+
+}
+
+export async function* hashtagsIterator(domain, hashtags) {
+    const buffer = []
+
+    const streamingSubscription = hashtagsStreamingObservable(domain, hashtags).subscribe({
+        next: value => buffer.push(value),
+        error: error => console.error(error),
+        complete: () => console.log('complete')
+    })
+
+    const timelineGenerator = hashtagsTimelineIterator(domain, hashtags)
+
+    try {
+        while (true) {
+            if (buffer.length > 0) {
+                yield buffer.pop()
+            } else {
+                yield (await timelineGenerator.next()).value
+            }
+        }
+    } finally {
+        streamingSubscription.unsubscribe()
+        timelineGenerator.return()
     }
 }
 
+
 const processStatus = (domain, status) => ({
     title: '',
     date: new Date(status.created_at),

src/services/misc.js

@@ -91,13 +91,17 @@ export async function* raceIterator(iterators) {
     }
 }
 
-export async function* tracksIterator(statusesIterator, cache) {
+export async function* tracksIterator(statusesGenerator, cache) {
-    yield* execPipe(
+    try {
-        statusesIterator,
+        yield* execPipe(
-        asyncFilter(track => track != null), // should not be necessary
+            statusesGenerator,
-        asyncFilter(notKnown(cache)),
+            asyncFilter(track => track != null), // should not be necessary
-        asyncMap(completeTrack)
+            asyncFilter(notKnown(cache)),
-    )
+            asyncMap(completeTrack)
+        )
+    } finally {
+        statusesGenerator.return()
+    }
 }
 
 const notKnown = cache => track => {