refact structure and api implem

.babelrc

@@ -1,7 +1,8 @@
 {
   "plugins": [
     ["@babel/plugin-transform-runtime", {
-      "corejs": 3
+      "corejs": 3,
+      "proposals": true
     }]
   ]
 }

package.json

@@ -19,7 +19,7 @@
     "svelte": "^3.16.7"
   },
   "dependencies": {
-    "@babel/runtime-corejs3": "^7.7.7",
+    "@babel/runtime-corejs3": "^7.8.3",
     "date-fns": "^2.9.0",
     "get-urls": "^9.2.0",
     "iter-tools": "^7.0.0-rc.0"

src/components/Viewer.svelte

@@ -22,13 +22,16 @@
 
     {#if duration}
         {currentTimeText}
+
         <input
             type="range"
             min="0"
             max={duration}
-            value={currentTime}
-            on:input={(e) => seek(e.target.value, false)}
-            on:change={(e) => seek(e.target.value, true)}>
+            value="0"
+            on:input={event => updateCurrentTime(event.target.value, false)}
+            on:change={event => updateCurrentTime(event.target.value, true)}
+        >
+
         {durationText}
     {/if}
 </div>
@@ -36,7 +39,7 @@
 <script>
     import { get } from 'svelte/store'
     import YoutubePlayer from '/components/YoutubePlayer'
-    import { secondsToElapsedTime } from '/util.js'
+    import { secondsToElapsedTime } from '/services/misc.js'
     import { paused, muted, volume, current, selectNext, loading } from '/store.js'
 
     let ready = null
@@ -52,6 +55,11 @@
     $: if (ended || error) {
         selectNext()
     }
+
+    const updateCurrentTime = (seconds, seekAhead) => {
+        seek(seconds, seekAhead)
+        currentTime = seconds
+    }
 </script>
 
 <style>

src/components/YoutubePlayer.svelte

@@ -2,11 +2,12 @@
 
 <script>
     import { onMount, onDestroy } from 'svelte'
-    import { loadIframeApi, STATUS } from '/youtube.js'
-    import { queue } from '/util.js'
+    import { loadIframeApi, STATE } from '/services/youtube.js'
+    import { queue } from '/services/misc.js'
 
     let element
     let player
+    let animationFrameId
 
     // output props
     export let ready = false
@@ -44,11 +45,11 @@
 
     const setPaused = paused => enqueue(player => {
         if (paused) {
-            if (player.getPlayerState() === STATUS.PLAYING) {
+            if (player.getPlayerState() === STATE.PLAYING) {
                 player.pauseVideo()
             }
         } else {
-            if (player.getPlayerState() !== STATUS.PLAYING) {
+            if (player.getPlayerState() !== STATE.PLAYING) {
                 player.playVideo()
             }
         }
@@ -66,45 +67,52 @@
         }
     })
 
-
     const setVolume = volume => enqueue(player => {
         player.setVolume(volume)
     })
 
-
     export const seek = (seconds, allowSeekAhead) => enqueue((player) => {
         player.seekTo(seconds, allowSeekAhead)
-        // currentTime = player.getCurrentTime()
     })
 
     onMount(() => {
         loadIframeApi().then(api => {
             element.id = Math.random().toString(16).slice(2, 8)
 
-            const onReady = (event) => {
-                run(event.target)
-
-                setInterval(() => {
-                    currentTime = event.target.getCurrentTime()
-                }, 1000)
+            const onReady = ({ target: player }) => {
+                run(player)
             }
 
-            const onStateChange = (event) => {
-                console.log('stateChange', event)
-
-                switch (event.data) {
-                    case STATUS.UNSTARTED:
+            const onStateChange = ({ data: state, target: player }) => {
+                switch (state) {
+                    case STATE.UNSTARTED:
                         ready = true
                         break
 
-                    case STATUS.PLAYING:
-                        duration = event.target.getDuration()
+                    case STATE.PLAYING:
+                        if (duration === null) {
+                            duration = player.getDuration()
+                        }
+
                         break
 
-                    case STATUS.ENDED:
+                    case STATE.ENDED:
                         ended = true
                         break
                 }
+
+                if (state === STATE.PLAYING) {
+                    const step = () => {
+                        currentTime = player.getCurrentTime()
+                        animationFrameId = requestAnimationFrame(step)
+                    }
+
+                    animationFrameId = requestAnimationFrame(step)
+                } else {
+                    if (animationFrameId) {
+                        cancelAnimationFrame(animationFrameId)
+                    }
+                }
             }
 
             const onError = () => {

src/services/mastodon.js

@@ -0,0 +1,73 @@
+import { observableToAsyncIterator } from '/services/misc.js'
+import 'core-js/es7/observable.js'
+
+const LINK_RE = /<(.+?)>; rel="(\w+)"/gi
+
+function parseLinkHeader(link) {
+    const links = {}
+
+    for (const [ , url, name ] of link.matchAll(LINK_RE)) {
+        links[name] = url
+    }
+
+    return links
+}
+
+export const hashtagStreamingObservable = (domain, hashtag) => {
+    return new Observable(observer => {
+        const onStatus = (event) => observer.next(JSON.parse(event.data))
+        const onError = (error) => observer.error(error)
+
+        const eventSource = new EventSource(`https://${domain}/api/v1/streaming/hashtag?tag=${hashtag}`)
+        eventSource.addEventListener('update', onStatus)
+        eventSource.addEventListener('error', onError)
+
+        return () => {
+            eventSource.removeEventListener('update', onStatus)
+            eventSource.removeEventListener('error', onError)
+        }
+    })
+}
+
+export async function* hashtagTimelineIterator (domain, hashtag) {
+    let nextLink = `https://${domain}/api/v1/timelines/tag/${hashtag}?limit=40`
+
+    while (nextLink) {
+        const response = await fetch(nextLink)
+
+        nextLink = response.headers.has('link')
+            ? parseLinkHeader(response.headers.get('link')).next
+            : null
+
+        yield* await response.json()
+    }
+}
+
+export async function* hashtagIterator(domain, hashtag) {
+    const newerIterator = observableToAsyncIterator(hashtagStreamingObservable(domain, hashtag))
+    const olderIterator = hashtagTimelineIterator(domain, hashtag)
+
+    let newer = newerIterator.next()
+    let older = olderIterator.next()
+
+    while (true) {
+        const promises = [newer, older].map((promise, index) => promise.then(result => ({ index, result })))
+        const { index, result: { done, value } } = await Promise.race(promises)
+
+        switch (index) {
+            default:
+                throw new Error()
+
+            case 0:
+                newer = newerIterator.next()
+                break;
+
+            case 1:
+                older = olderIterator.next()
+                break;
+        }
+
+        yield value
+    }
+}
+

src/services/misc.js

@@ -0,0 +1,139 @@
+import getUrls from 'get-urls'
+import { execPipe, asyncFilter, asyncMap } from 'iter-tools'
+
+export const tap = f => x => {
+    f(x)
+    return x
+}
+
+export const queue = () => {
+    const deferred = defer()
+    let promise = deferred.promise
+
+    const enqueue = f => {
+        promise = promise.then(tap(f))
+    }
+
+    return { enqueue, run: deferred.resolve }
+}
+
+export const defer = () => {
+    let resolve
+    let reject
+
+    const promise = new Promise((res, rej) => {
+        resolve = res
+        reject = rej
+    })
+
+    return { resolve, reject, promise }
+}
+
+export async function* observableToAsyncIterator(observable) {
+    const buffer = [defer()]
+    let done = false
+
+    const next = (x) => {
+        buffer[buffer.length - 1].resolve(x)
+        buffer.push(defer())
+    }
+
+    const error = (error) => {
+        buffer[buffer.length - 1].reject(error)
+    }
+
+    const complete = (x) => {
+        buffer[buffer.length - 1].resolve(x)
+        done = true
+    }
+
+    const subscription = observable.subscribe({ next, error, complete })
+
+    try {
+        while (true) {
+            const value = yield await buffer[0].promise
+            buffer.unshift()
+
+            if (done) {
+                return value
+            } else {
+                yield value
+            }
+        }
+    } finally {
+        subscription.unsubscribe()
+    }
+}
+
+export function intersection(xs, ys) {
+    return xs.filter(x => ys.includes(x))
+}
+
+export const secondsToElapsedTime = (seconds) => {
+    const parts = [
+        Math.floor(seconds / 3600),
+        Math.floor(seconds / 60) % 60,
+        Math.floor(seconds) % 60
+    ]
+
+    return parts
+        .filter((value, index) => value > 0 || index > 0)
+        .map(value => value < 10 ? '0' + value : value)
+        .join(':')
+}
+
+export async function* mkTracksIterator(statusesIterator) {
+    // const known = new Set()
+    const knownStatus = {}
+    const knownYoutube = {}
+
+    const tracks = execPipe(
+        statusesIterator,
+        asyncFilter(status => {
+            if (knownStatus.hasOwnProperty(status.id)) {
+                return false
+            } else {
+                knownStatus[status.id] = null
+                return true
+            }
+        }),
+        asyncMap(status => ({ status, data: mkData(status) })),
+        asyncFilter(({ data }) => {
+            if (!data) {
+                return false
+            } else {
+                if (knownYoutube.hasOwnProperty(data.id)) {
+                    return false
+                } else {
+                    knownYoutube[data.id] = null
+                    return true
+                }
+            }
+        }),
+        asyncMap(async ({ status, data }) => ({ status, data, metadata: await mkMetadata(data) }))
+    )
+
+    yield* tracks
+}
+
+function mkData(status)
+{
+    const urls = getUrls(status.content)
+
+    for (const urlAsString of urls) {
+        const url = new URL(urlAsString)
+
+        if (['youtube.com', 'music.youtube.com'].includes(url.hostname) && url.searchParams.has('v')) {
+            return { url: urlAsString, id: url.searchParams.get('v') }
+        } else if (url.hostname === 'youtu.be') {
+            return { url: urlAsString, id: url.pathname.substring(1) }
+        }
+    }
+
+    return null
+}
+
+async function mkMetadata(entry) {
+    return fetch(`https://noembed.com/embed?url=https://www.youtube.com/watch?v=${entry.id}`)
+        .then(response => response.json())
+}

src/services/svelte.js

@@ -0,0 +1,10 @@
+import { writable } from 'svelte/store'
+
+export const writableLocalStorage = (key, value) => {
+    const item = JSON.parse(localStorage.getItem(key))
+    const store = writable(item === null ? value : item)
+
+    store.subscribe(x => localStorage.setItem(key, JSON.stringify(x)))
+
+    return store
+}

src/services/youtube.js

@@ -1,6 +1,6 @@
 const IFRAME_API_URL = 'https://www.youtube.com/iframe_api'
 
-export const STATUS = {
+export const STATE = {
     UNSTARTED: -1,
     ENDED: 0,
     PLAYING: 1,
@@ -15,6 +15,7 @@ const loadScript = (attributes) => {
             throw new Error('src is required')
         }
 
+        // // we could optimize futher by checking if a script with iframe api as src is already loading
         // const scripts = Array.from(document.getElementsByTagName('script'))
         //     .filter(script => script.src === attribute.src)
 

src/store.js

@@ -1,5 +1,7 @@
 import { writable, derived, get } from 'svelte/store'
-import { writableLocalStorage, mkTracksIterator } from '/util.js'
+import { writableLocalStorage } from '/services/svelte.js'
+import { hashtagIterator } from '/services/mastodon.js'
+import { mkTracksIterator } from '/services/misc.js'
 
 export const domain = writableLocalStorage('domain', 'eldritch.cafe')
 
@@ -26,11 +28,14 @@ export const loading = writable(false)
 
 
 
-const tracksIterator = mkTracksIterator(get(domain), get(hashtags))
+
 
 export const selectPrevious = () => { if (get(canPrevious)) index.update($index => $index - 1) }
 export const selectNext = () => { if (get(canNext)) index.update($index => $index + 1) }
 
+
+const tracksIterator = mkTracksIterator(hashtagIterator(get(domain), get(hashtags)[0]))
+
 export const enqueue = async () => {
     if (!get(enqueueing)) {
         enqueueing.set(true)

src/util.js

@@ -1,171 +0,0 @@
-import { writable } from 'svelte/store'
-import getUrls from 'get-urls'
-import { execPipe, asyncFilter, asyncMap } from 'iter-tools'
-
-export const tap = f => x => {
-    f(x)
-    return x
-}
-
-export const queue = () => {
-    const deferred = defer()
-    let promise = deferred.promise
-
-    const enqueue = f => {
-        promise = promise.then(tap(f))
-    }
-
-    return { enqueue, run: deferred.resolve }
-}
-
-export const defer = () => {
-    let resolve
-    let reject
-
-    const promise = new Promise((res, rej) => {
-        resolve = res
-        reject = rej
-    })
-
-    return { resolve, reject, promise }
-}
-
-export const writableLocalStorage = (key, value) => {
-    const item = JSON.parse(localStorage.getItem(key))
-    const store = writable(item === null ? value : item)
-
-    store.subscribe(x => localStorage.setItem(key, JSON.stringify(x)))
-
-    return store
-}
-
-const millisecond = 1
-const second = 1000 * millisecond
-const minute = 60 * second
-const hour = 60 * minute
-
-export const secondsToElapsedTime = (seconds) => {
-    const parts = [
-        Math.floor(seconds / 3600),
-        Math.floor(seconds / 60) % 60,
-        Math.floor(seconds) % 60
-    ]
-
-    return parts
-        .filter((value, index) => value > 0 || index > 0)
-        .map(value => value < 10 ? '0' + value : value)
-        .join(':')
-}
-
-export async function* mkStatusesIterator(domain, hashtag) {
-    console.log(`Initialize statuses iterator for #${hashtag} on ${domain}`)
-    const buffer = []
-
-
-
-    // streaming
-    const eventSource = new EventSource(`https://${domain}/api/v1/streaming/hashtag?tag=${hashtag}`)
-
-    eventSource.addEventListener('update', (e) => {
-        console.log(`Received new recent status for #${hashtag} on ${domain}`)
-        buffer.unshift(JSON.parse(e.data))
-    })
-
-    eventSource.onerror = (error) => console.log('onerror', error)
-
-
-
-    // timeline
-    let nextLink = `https://${domain}/api/v1/timelines/tag/${hashtag}?limit=40`
-
-    while (true) {
-        if (buffer.length === 0) {
-            console.log(`Fetch timeline for #${hashtag} on ${domain}`)
-            const next = await fetchTimeline(nextLink)
-
-            if (next.statuses.length) {
-                buffer.push(...next.statuses)
-                nextLink = next.links.next
-            }
-        }
-
-        yield buffer.shift()
-    }
-}
-
-export async function* mkTracksIterator(domain, hashtags) {
-    // const known = new Set()
-    const known = {}
-    const [hashtag] = hashtags
-
-    const statuses = mkStatusesIterator(domain, hashtag)
-
-    const tracks = execPipe(
-        statuses,
-        asyncMap(status => ({ status, data: mkData(status) })),
-        asyncFilter(({ data }) => {
-            if (data) {
-                // const found = known.has(data.id)
-                // known.add(data.id)
-                const found = known.hasOwnProperty(data.id)
-                known[data.id] = true
-                return !found
-            }
-
-            return false
-        }),
-        asyncMap(async ({ status, data }) => ({ status, data, metadata: await mkMetadata(data) }))
-    )
-
-    yield* tracks
-}
-
-export async function fetchTimeline(url) {
-    console.log(`fetching ${url}`)
-    const response = await fetch(url)
-    const statuses = await response.json()
-
-    const links = response.headers.has('link')
-        ? parseLinkHeader(response.headers.get('link'))
-        : {}
-
-    return { statuses, links }
-}
-
-const LINK_RE = /<(.+?)>; rel="(\w+)"/gi
-
-function parseLinkHeader(link) {
-    const links = {}
-
-    for (const [ , url, name ] of link.matchAll(LINK_RE)) {
-        links[name] = url
-    }
-
-    return links
-}
-
-function mkData(status)
-{
-    const urls = getUrls(status.content)
-
-    for (const urlAsString of urls) {
-        const url = new URL(urlAsString)
-
-        if (['youtube.com', 'music.youtube.com'].includes(url.hostname) && url.searchParams.has('v')) {
-            return { url: urlAsString, id: url.searchParams.get('v') }
-        } else if (url.hostname === 'youtu.be') {
-            return { url: urlAsString, id: url.pathname.substring(1) }
-        }
-    }
-
-    return null
-}
-
-async function mkMetadata(entry) {
-    return fetch(`https://noembed.com/embed?url=https://www.youtube.com/watch?v=${entry.id}`)
-        .then(response => response.json())
-}
-
-export function intersection(xs, ys) {
-    return xs.filter(x => ys.includes(x))
-}