feat: streaming feed #4

molekilla · 2022-07-05T16:56:35Z

Pending docs

nugaon

There is a missing method in the implementation setUpdate and also presumably a wrong implementation of getUpdates. Please correct these and also refactor completely the docs written based on my comments.

nugaon · 2022-07-11T08:27:49Z

README.md

+import { Bee, Reference, Topic, Utils } from '@ethersphere/bee-js'
+import type { HexString } from '@ethersphere/bee-js/dist/src/utils/hex'
+import { SequentialFeed } from '../../src/sequential-feed'
+import { assertBytes, Bytes, bytesToHex, hexToBytes, makePrivateKeySigner } from '../../src/utils'
+import { beeUrl, getPostageBatch } from '../utils'
+
+const myIdentity = {
+    privateKey: '...private key as hex...' as HexString,
+    publicKey: '03c32bb011339667a487b6c1c35061f15f7edc36aa9a0f8648aba07a4b8bd741b4' as HexString,
+    address: '8d3766440f0d7b949a5e32995d09619a7f86e632' as HexString,
+}
+const owner = Utils.Hex.makeHexString(myIdentity.address, 40)
+const signer = makePrivateKeySigner(hexToBytes(myIdentity.privateKey) as Bytes<32>)
+const topic = '0000000000000000000000000000000000000000000000000000000000000000' as Topic
+const bee = new Bee(beeUrl())
+const batchId = getPostageBatch()


this sample code with relative imports is wrong as you cannot run the script anywhere.
a sample code should work in a clean starting environment and/or that has to be simple (few lines of clean code).

This sample code does not satisfy either one above.

nugaon · 2022-07-11T09:22:59Z

src/streaming.ts

+  writeUint64BigEndian,
+} from './utils'
+
+export type StreamingFeedData = {


this is basically StreamingFeedChunk but index is chunkIndex.
Delete it and use `StreamingFeedChunk only

nugaon · 2022-07-11T09:24:51Z

src/streaming.ts

+}
+
+/** Swarm Feed Read and Write operations */
+export interface SwarmStreamingFeedRW<Index = number> extends SwarmStreamingFeedR {


you don't use your declared type variable.

Suggested change

export interface SwarmStreamingFeedRW<Index = number> extends SwarmStreamingFeedR {

export interface SwarmStreamingFeedRW extends SwarmStreamingFeedR<number> {

also, setUpdate is missing.

nugaon · 2022-07-11T09:32:31Z

src/streaming.ts

+export interface SwarmStreamingFeedR<Index = number> extends SwarmFeedHandler {
+  getIndexForArbitraryTime(lookupTime: number, initialTime?: number, updatePeriod?: number): Promise<Index> | Index
+  getUpdate(initialTime: number, updatePeriod: number, lookupTime?: Index): Promise<StreamingFeedChunk<Index>>
+  getUpdates(initialTime: number, updatePeriod: number): Promise<StreamingFeedChunk<Index>[]>


would be nice to pass a time interval for fetching e.g. lookupTimeInterval: [number, number]

nugaon · 2022-07-11T09:33:28Z

src/streaming.ts

+
+/** Swarm Feed Read operations */
+export interface SwarmStreamingFeedR<Index = number> extends SwarmFeedHandler {
+  getIndexForArbitraryTime(lookupTime: number, initialTime?: number, updatePeriod?: number): Promise<Index> | Index


it shouldn't give back Promise as it can be calculated completely client-side

nugaon · 2022-07-11T10:49:47Z

src/streaming-feed.ts

+      const index = await getIndexForArbitraryTime(getCurrentTime(), initialTime, updatePeriod)
+      const socChunk = await socReader.download(this.getIdentifier(topicBytes, index - 1))
+
+      let feed = await mapSocToFeed(socChunk)


not necessary await

nugaon · 2022-07-11T10:51:32Z

src/streaming-feed.ts

+      const socChunk = await socReader.download(this.getIdentifier(topicBytes, index - 1))
+
+      let feed = await mapSocToFeed(socChunk)
+      let i = 1


the feed indexing starts at 0 as far as I know.

nugaon · 2022-07-11T10:56:07Z

src/streaming-feed.ts

+      // while from last to first, use lookupTime = chunk.timestamp + 1
+
+      const index = await getIndexForArbitraryTime(getCurrentTime(), initialTime, updatePeriod)
+      const socChunk = await socReader.download(this.getIdentifier(topicBytes, index - 1))


why you download the penultimate chunk instead of the last one?

nugaon · 2022-07-11T11:00:46Z

src/streaming-feed.ts

+      let i = 1
+      feeds.push(feed)
+      while (i < index) {
+        feed = await getUpdate(initialTime, updatePeriod, feed.timestamp + updatePeriod + 1)


I cannot comprehend with this section here.
you downloaded the penultimate chunk from which you go forward in updates that you repeat times based on the index of the penultimate chunk...
can you elaborate please? maybe I'm missing something.

nugaon · 2022-07-11T11:03:16Z

src/streaming-feed.ts

+  }
+
+  /** Get Single Owner Chunk identifier */
+  public getIdentifier(topic: Utils.Bytes.Bytes<32>, index: number): Utils.Bytes.Bytes<32> {


please provide detailed JSDoc here, since index: number is disambiguous because timestamp is also number as well as its transformed sequential index.

crtahlin · 2022-08-11T11:23:17Z

Is this done, I guess it is ready for review @molekilla ?

nugaon · 2022-08-18T11:39:55Z

src/sequential-feed.ts

+  /**
+   * Creates a sequential feed reader
+   * @param topic a swarm topic
+   * @param signer signer


there is no such a parameter.

nugaon · 2022-08-18T11:42:47Z

src/sequential-feed.ts

+    /**
+     * Gets the last index in the feed
+     * @returns An index number
+     */


It does have any sense to define JSDoc above this, and the other the arrow functions. if you want to keep these then define these about the corresponding interface definitions.

I will fix the sequential feed, remember it was started by you :)

nugaon · 2022-08-18T11:45:21Z

src/streaming.ts

+}
+
+/** Interface for feed type classes */
+export interface SwarmStreamingFeed<Index> {


please, rename it to IStreamingFeed

nugaon · 2022-08-18T11:45:56Z

src/streaming.ts

+/** Interface for feed type classes */
+export interface SwarmStreamingFeed<Index> {
+  /** Feed type identifier */
+  readonly type: FeedType


feed type here is only fault-tolerant-stream

nugaon · 2022-08-18T11:51:17Z

src/streaming.ts

+}
+
+/** Swarm Feed Read operations */
+export interface SwarmStreamingFeedR extends SwarmFeedHandler {


the already defined SwarmFeedR why was not good enough?

findLastUpdate and getLastIndex is missing

SwarmStreamingFeedR I will leave it, streaming uses the index in every method call and before doing that changed I circled back with you about it. Let me know.

Will add the other two

nugaon · 2022-08-18T11:57:11Z

src/streaming.ts

+  const updatePeriod = writeUint64BigEndian(options.updatePeriod)
+  const chunkIndex = writeUint64BigEndian(options.index)
+
+  return serializeBytes(chunkIndex, updatePeriod, timestamp, reference)


I suggest that you should check here how long the reference is and encode it to the soc payload.

nugaon · 2022-08-18T12:22:27Z

test/integration/sequential-feed.spec.ts

@@ -39,10 +39,12 @@ describe('feed', () => {
  }, 21000)

  test('multiple updates using setUpdate and lookup', async () => {
-    const reference = Utils.Hex.makeHexString('0000000000000000000000000000000000000000000000000000000000000000', 64)
+    const reference = Utils.Hex.makeHexString(new Date().getTime().toString().padStart(64, '0'), 64)


why did you touch this part?

it is not related to the streaming of feeds

it messes up the inteded test logic as you check the code below.

nugaon · 2022-08-18T12:22:53Z

test/integration/streaming-feed.spec.ts

+  test('lookup for empty feed update', async () => {
+    const emptyTopic = '1200000000000000000000000000000000000000000000000000000000000001' as Topic
+    const feedR = streamingFeed.makeFeedR(emptyTopic, testIdentity.address)
+    const lastIndex = await feedR.getIndexForArbitraryTime(getCurrentTime())


unnecesarry await

nugaon · 2022-08-18T12:23:09Z

test/integration/streaming-feed.spec.ts

+    const initialTime = getCurrentTime()
+
+    const sleep = async (seconds: number) =>
+      new Promise((resolve, reject) => {


not used reject variable.

nugaon · 2022-08-18T12:24:39Z

src/streaming-feed.ts

+     * @param lookupTime lookup time
+     * @returns Returns -1 if not found, otherwise the index
+     */
+    const getIndexForArbitraryTime = (lookupTime: number, initialTime: number, updatePeriod: number): number => {


I would move it to a class method or normal helper function.

feat(feed): streaming feed

9c45d4a

molekilla requested a review from nugaon July 5, 2022 16:56

feat(feed): docs

948fc5c

nugaon requested changes Jul 11, 2022

View reviewed changes

molekilla added 3 commits July 11, 2022 18:03

feat: requested changes from code review, pending jsdoc

1f1a503

docs: added jsdoc

0f11f46

docs: updated readme

727bb1b

molekilla requested a review from nugaon July 12, 2022 22:12

molekilla added 4 commits July 21, 2022 07:13

ci: testing node

2654a5b

ci: updates

f8b6a1b

ci: use fdp play

a1a96b1

ci: bee 1.6.2

f80cf35

crtahlin mentioned this pull request Aug 10, 2022

Implement Fault Tolarent Feed Streams #3

Open

crtahlin added the enhancement New feature or request label Aug 11, 2022

crtahlin assigned molekilla Aug 11, 2022

nugaon requested changes Aug 18, 2022

View reviewed changes

molekilla added 2 commits August 18, 2022 12:19

feat: resolving some code review comments

ddd52a1

feat: adds getLastIndex and findLastUpdate

95cf561

molekilla mentioned this pull request Aug 23, 2022

feat: json #2

Open

nugaon added 5 commits August 15, 2023 16:51

refactor: refactor

d6309fa

fix: time resolution

36623ad

revert: sec to ms

2b2b629

refactor: make getLastIndex synchronous

a224273

fix: discover updates

327c39b

nugaon changed the title ~~feat(feed): streaming feed~~ feat: streaming feed Aug 21, 2023

nugaon added 4 commits August 21, 2023 18:04

docs: fix

9858cb2

ci: remove node 14 since it is not supported anymore

53f7a2e

docs: bump node version in the badge

e4a3a53

test: added reference check

4500ee5

test: add irregular update tests

4180f4b

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: streaming feed #4

feat: streaming feed #4

molekilla commented Jul 5, 2022

nugaon left a comment

nugaon Jul 11, 2022

nugaon Jul 11, 2022

nugaon Jul 11, 2022

nugaon Jul 11, 2022

nugaon Jul 11, 2022

nugaon Jul 11, 2022

nugaon Jul 11, 2022

nugaon Jul 11, 2022

nugaon Jul 11, 2022

nugaon Jul 11, 2022

nugaon Jul 11, 2022

crtahlin commented Aug 11, 2022

nugaon Aug 18, 2022

nugaon Aug 18, 2022

molekilla Aug 18, 2022

nugaon Aug 18, 2022

nugaon Aug 18, 2022

nugaon Aug 18, 2022

nugaon Aug 18, 2022

molekilla Aug 18, 2022

nugaon Aug 18, 2022

nugaon Aug 18, 2022

nugaon Aug 18, 2022

nugaon Aug 18, 2022

nugaon Aug 18, 2022

	export interface SwarmStreamingFeedRW<Index = number> extends SwarmStreamingFeedR {
	export interface SwarmStreamingFeedRW extends SwarmStreamingFeedR<number> {

feat: streaming feed #4

Are you sure you want to change the base?

feat: streaming feed #4

Conversation

molekilla commented Jul 5, 2022

nugaon left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

crtahlin commented Aug 11, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment