feat: initial subscriptions implementation by darahayes · Pull Request #32 · aerogear-attic/data-sync-server

darahayes · 2018-07-23T22:32:25Z

To do

Tests
Cleanup???

This is runnable right now though. You can try it out as follows:

npm run db:init:memeo
npm run dev
Open http://localhost:8000/graphiql in two different tabs
Leave one tab as it is with the sample queries
In the other tab remove the sample queries and paste the following:

subscription memeAdded{
  memeAdded{
    id
    photoUrl
  }
}

Run the subscription query and you should see the following message: "Your subscription data will appear here after server publication!"
Run the createMeme query in the other window.

How does the subscription piece work?

On the server side we define a new memeAdded subscription.

type Subscription {
    memeAdded(photoUrl: String):Meme!
}

Clients will be able to use this query to listen for updates. The photoUrl variable is optional (more on that later).

The next thing we need to be able to do is define a way to publish an event during some query or mutation. GraphQL uses a Pub/Sub model to do this. So when a query/mutation occurs, we can publish some events+data to a Pub/Sub mechanism and we can react to these events elsewhere. Below shows how a user could configure a createMeme resolver to also publish an event.

{
    type: 'Mutation',
    field: 'createMeme',
    requestMapping: `{
      "operation": "insert",
      "doc": {
        "_type":"meme",
        "photoUrl": "{{context.arguments.photoUrl}}"
      }
    }`,
    responseMapping: '{{ toJSON (convertNeDBIds context.result) }}',
    publish: JSON.stringify({
      topic: 'memeCreated',
      payload: `{
        "memeAdded": {{ toJSON context.result }},
      }`
    }),
  }

The last piece of the puzzle is to be able to listen for those events and push them back down to a client. We can do this with the new Subscription config object. Take a look at the example one for the memeolist (included in this PR)

{
    type: 'Subscription',
    field: 'memeAdded',
    topic: 'memeCreated',
}

This config tells our server to listen for messages published to the 'memeCreated' topic on the internal pubsub mechanism and to fire the memeAdded resolver as a result.

Trying out Filtering

Filtering is an important aspect of Grapqhl subscriptions. In most real world apps, it doesn't make sense for a client to be listening for notifications about every single record that was added to a collection. It would be like getting a notification in slack every time anybody in the world sent a message.

Taking the Slack example further, on the client side we might subscribe to a messageAdded query as follows:

subscription messageAdded{
  messageAdded(channelId: '#random'){
    id
	channelId
    text
	userId
  }
}

The idea is we only want to listen for notifications in a particular slack channel - #random.

How would this look on the server side?

How can we enable an end user of our service to define some kind of filtering that's relatively good?

I have come up with small DSL that allows the user to define some conditions in a declarative way using JSON. We add this filter to the subscription config object.

Here's when it might look like in the Slack example

{
    type: 'Subscription',
    field: 'messageAdded',
    topic: 'messageCreated',
	filter: JSON.stringify({
      'eq': ['$payload.messageAdded.channelId', '$variables.channelId']
    })
  }

This filter would check that the channelId in the newly created message is equal to the channelId that the client passed as a variable when they subscribed. i.e. the user can say give me real time updates about messages for this particular channel.

The filter expression has access to a context object that looks like this: { payload: {...}, variables: {}} where payload is the message received from the pubsub layer and variables contains whatever variables were passed by the client to initialize the subscription. Using the dollar syntax like $payload.messageAdded.id we can reference the context object.

I figured that 99% of the time, the basic use case will like the one above where we check that two things are equal. I made a more concise syntax for this particular case. The same example above can be expressed as follows:

{
    type: 'Subscription',
    field: 'messageAdded',
    topic: 'messageCreated',
	filter: JSON.stringify({
      '$payload.messageAdded.channelId': '$variables.channelId'
    })
  }

This is the most basic usage of the filter language. You can do many more things. Try some of these out with the memeolist schema:

(you can modify the subscription object in the file located under seeders, then run the db init and start commands again)

// subscribe only to new memes where payload.photoURL === google.com
{
    type: 'Subscription',
    field: 'memeAdded',
    GraphQLSchemaId: 2,
    topic: 'memeCreated',
    createdAt: time,
    updatedAt: time,
	filter: JSON.stringify({
      'eq': ['$payload.memeAdded.photoUrl', 'https://google.com']
    })
  }

// subscribe only to new memes where payload.photoURL !== google.com
{
    type: 'Subscription',
    field: 'memeAdded',
    GraphQLSchemaId: 2,
    topic: 'memeCreated',
    createdAt: time,
    updatedAt: time,
	filter: JSON.stringify({
      '!eq': ['$payload.memeAdded.photoUrl', 'https://google.com']
    })
  }

// subscribe only to new memes where payload.photoURL matches a regex
{
    type: 'Subscription',
    field: 'memeAdded',
    GraphQLSchemaId: 2,
    topic: 'memeCreated',
    createdAt: time,
    updatedAt: time,
	filter: JSON.stringify({
      'match': ['$payload.memeAdded.photoUrl', 'https:\/\/.*']
    })
  }

// subscribe only to new memes where payload.photoUrl === https://google.com || payload.photoUrl == $variables.photoUrl
{
    type: 'Subscription',
    field: 'memeAdded',
    GraphQLSchemaId: 2,
    topic: 'memeCreated',
    createdAt: time,
    updatedAt: time,
	filter: JSON.stringify({
      'or': [
        { 'eq': ['$payload.memeAdded.photoUrl', 'https://google.com'] },
        { 'eq': ['$payload.memeAdded.photoUrl', '$variables.photoUrl']}
       ]
    })
  }

// subscribe to new memes where payload.photoUrl === variables.photoUrl || (payload.photoUrl === https://facebook.com && payload.photoUrl === Dara)
{
    type: 'Subscription',
    field: 'memeAdded',
    GraphQLSchemaId: 2,
    topic: 'memeCreated',
    createdAt: time,
    updatedAt: time,
	filter: JSON.stringify({
      'or': [
        { 'eq': ['$payload.memeAdded.photoUrl', '$variables.photoUrl']},
        { 'and': [
                  { 'eq': ['$payload.memeAdded.photoUrl', 'https://facebook.com']},
                  { 'eq': ['$payload.memeAdded.owner', 'Dara'] }
                ]
       ]
    })
  }

Why Did I do it this way?

JSON format is super portable, for example - instead of defining the filter expression in the server config, a client app could define whatever filter they want and use it to listen for updates, and it would be completely safe because there's no way for them to inject code. This has the potential for some very useful applications IMO.
I considered using handlebars to do something similar but it's very hacky because it doesn't actually evaluate things to true or false. It would also mean users would have to write a horrible syntax. In my opinion, this syntax is pretty intuitive.
Inspired by MongoDB query language which I thought was very intuitive
Flexibility, the code that evaluates these expressions could be extended in the future very easily.

darahayes · 2018-07-23T22:38:36Z

@@ -0,0 +1,91 @@
+const _ = require('lodash')
+
+module.exports = function ExpressionParser () {


This is actually really generic and could be published as its own npm module with docs that demonstrate usage.

I think it might be a better idea to extract this.

BTW, have you searched for a lib that does the same thing?

I tried to search for something like this but I couldn't find anything. What would you even call this? I don't even know how to describe it in 2-3 words so I found it very hard to search for.

I think it would be great to extract. We should be constantly looking for opportunities to extract and open source internal libs like this. Where should the repo go? under github.com/aerogear?

wtrocki · 2018-07-24T07:05:30Z


  type Subscription {
-    _: Boolean
+    memeAdded(photoUrl: String):Meme!


[Outside scope of this Pr] I think it will be more benefitial to add more fields to the meme like votes as this will help us to test various use cases.

aliok · 2018-07-24T08:12:20Z


+  // Set up the WebSocket for handling GraphQL subscriptions
+  /* eslint-disable no-new */
+  new SubscriptionServer({


this won't pick up new schema with hot reload

👍 This completely slipped my mind

aliok · 2018-07-24T08:16:30Z

-  let dataSourcesJson = dataSources.map((dataSource) => {
-    return dataSource.toJSON()
-  })
+  let dataSourcesJson = await models.DataSource.findAll({raw: true})


aliok · 2018-07-24T08:20:32Z

+const PubSub = require('./pubsubNotifiers').InMemory
+
+module.exports = function (schemaString, dataSourcesJson, resolverMappingsJson, subscriptionMappingsJson) {
+  const pubsub = PubSub()


what's the plan to abstract this away?

aliok · 2018-07-24T08:27:19Z

+    if (typeof str !== 'string' || str.length <= 1) {
+      return str
+    }
+    var c1 = str[0]


I see some vars in the file

aliok · 2018-07-24T08:33:06Z

+    return new Promise(async (resolve, reject) => {
+      try {
+        const result = await resolverFn(obj, args, context, info)
+        resolve(result)


+1 on resolving and then publishing

aliok · 2018-07-24T08:34:50Z

+        let payload = compiledPayload(compileOpts)
+
+        // The InMemory pubsub implementation wants an object
+        // Whereas the postgres one would expect a string


as you noted, we need abstraction for this. let's not forget about that.

aliok · 2018-07-24T08:38:23Z

@@ -0,0 +1,91 @@
+const _ = require('lodash')
+
+module.exports = function ExpressionParser () {


I think it might be a better idea to extract this.

BTW, have you searched for a lib that does the same thing?

aliok · 2018-07-24T08:44:44Z

+      payload: `{
+        "memeAdded": {{ toJSON context.result }},
+      }`
+    }),


can you also add sample into memeolist.query.graphql file so that we don't have to type anything when we open Graphiql?

aliok · 2018-07-24T08:47:59Z

@darahayes really great work! and, thanks for the long PR description which explains everything!
left some comments, but won't affect the general structure.

I haven't done any verification yet. I will do it after the requested changes are addressed.

I think we would need some integration tests where we create 2 clients and mutate on one and listen on other one. Feel free to implement it as part of this pr OR we can implement it in a separate ticket/pr.
UPDATE: on a second thought, better to do it in another ticket to unblock merging of this PR. created https://issues.jboss.org/browse/AEROGEAR-7691

Another thing: we need to discuss with David and the team about syncing the sync server instances. I know we talked about it and it seems necessary, but maybe it is not in the scope right now.

wtrocki · 2018-07-24T09:03:59Z

+const { evalWithContext } = require('../util/filterEvaluator')()
+const { log } = require('../util/logger')
+
+module.exports = function mapSubscriptions (subsciptionMappings, pubsub) {


I would add some comments on what this module is doing.

Great suggestion

wtrocki · 2018-07-24T09:06:49Z

+        const result = await resolverFn(obj, args, context, info)
+        resolve(result)
+
+        const { pubsub, topic, compiledPayload } = publishOpts


I would wrap the code from here into separate function like notifySubscriptions() that could be used in different context.

wtrocki · 2018-07-24T09:11:27Z

@darahayes Performing verification and adjusting memelist app

wtrocki · 2018-07-24T10:31:48Z

@darahayes Adjusting now the memolist app for testing. Initial subscription looked like this. Do we support action filter from mobile? Something like here: https://github.com/aerogear/aerogear-android-example-apps/blob/07fcffdb17b4a7bd2543097b1b5e8a6aef159dc5/memeolist/app/src/main/graphql/org/aerogear/android/app/memeolist/graphql/memes.graphql#L17-L19

darahayes · 2018-07-24T11:02:53Z

@wtrocki I have never seen this kind of syntax before. We definitely don't support it right now. Can you link to some Apollo/Gprahql documentation that shows this off in more detail? I'd like to learn more

aliok · 2018-07-24T11:21:49Z

@darahayes @wtrocki that I think is from the schema generated by Graphcool. When you define a type it generates queries/mutations/subscriptions.
Example type:

type Note {
  id: ID! @isUnique
  text: String
}

Portion of generated stuff:

allNotes(filter: NoteFilter, orderBy: NoteOrderBy, skip: Int, after: String, before: String, first: Int, last: Int): NoteConnection!
...
input NoteFilter {
  # Logical AND on all given filters.
  AND: [NoteFilter!]

  # Logical OR on all given filters.
  OR: [NoteFilter!]
  id: ID

  # All values that are not equal to given value.
  id_not: ID

  # All values that are contained in given list.
  id_in: [ID!]

  # All values that are not contained in given list.
  id_not_in: [ID!]
...
}

That's still valid Graphql.

wtrocki · 2018-07-24T11:48:17Z

Verified with graphql. Verified on the Android application.
Rebase needed! I will be nice to put some of the description from this PR into some rough docs in the repository so we will have some documentation draft.

david-martin

Verified from a functionality point of view.
Structure of code looks very reasonable too. Good to have the wrapper around postgres pubsub
Approved.

darahayes · 2018-07-24T15:54:42Z

@aliok Thanks for the great feedback. I believe I have incorporated all of it except the thing about extracting the filterEvaluator library. That will require a lot of extra time.

Let me know what you think!

wtrocki · 2018-07-24T16:17:28Z

      log.info('Received schema change notification. Rebuilding it')
      let newSchema
      try {
+        newSubScriptionServer(server, schema)


Should this be the newSchema that is build bellow?

camilamacedo86 · 2018-07-24T16:33:33Z

+    throw new Error(`PubSub implementation for ${pubsubConfig.type} is missing a constructor`)
+  }
+
+  const pubsub = new PubSubClass(pubsubConfig.config)


Is not better add all const on top of it? All together.

@camilamacedo86 not sure what you mean, it wouldn't be possible to declare this at an earlier point in the code.

I think it should be inside of the func getPublish (PubSubClass); as described in the next comment.

camilamacedo86 · 2018-07-24T16:40:08Z

+module.exports = function NewPubSub (pubsubConfig) {
+  const PubSubClass = notifiers[pubsubConfig.type]
+
+  if (!PubSubClass) {


Is not better encapsulate these if in functions the call them, something like as:

module.exports = function NewPubSub (pubsubConfig) { const PubSubClass = notifiers[pubsubConfig.type] try { checkObject(PubSubClass); return getPublish(PubSubClass); } catch (error) { throws error; } }

Hi @camilamacedo86 thanks for reviewing! Your suggestion is very good and fair, but I do not see the need to move any of this stuff into separate functions right now because because this is the only place we will ever run these checks.

@darahayes the problem is if we don't start the code as it should be then it will not grow as we wish. This code has many ifs, and if someone will improve it will add more ifs then if we encapsulate the logic is not required comments because what it is doing will be clear as we are doing things that allow improvements in a healthy way. Is it makes sense for you?

camilamacedo86 · 2018-07-24T20:08:52Z

  }
 }

+const port = process.env.HTTP_PORT || '8000'


Should not better add all const/var declarations in the top without space?

Variable declarations at the top are typically used for declaring imports.

@darahayes at on top typically is declared all that is global to the program in all languages. For default, it follows the following standard/convention.

imports ... global vars ... functions/methods ...

It is something important to make easier to keep the project maintained, global vars in many places make harder to read and understand the code as could cause mistakes, for example, declare the same value again in another global var.

Note that this standard is valid for any language. global variables are often available by declaring a variable at the top level of the program. See here

Also, it is a good practice to avoid global variables in JS and try to use always local variable instead of it. Following some references.

https://www.w3schools.com/js/js_best_practices.asp

https://www.w3.org/wiki/JavaScript_best_practices

camilamacedo86 · 2018-07-24T20:10:54Z

+    subscriptionsEndpoint: `ws://${hostname()}:${port}/subscriptions`
  },
  postgresConfig: {
    database: process.env.POSTGRES_DATABASE || 'aerogear_data_sync_db',


IHMO the fixed values as 'aerogear_data_sync_db' could be declared as const on to of it.

+1 on that. @camilamacedo86 great suggestion. I will create a GH issue for that.
I don't want this PR to get any bigger. I am too lazy to create a JIRA for something minor though.

aliok · 2018-07-26T06:56:06Z

@darahayes verified functionality locally but only the case in memeo list schema. unit tests look good as well.
Could you please review the test cases at the integration tests JIRA? https://issues.jboss.org/browse/AEROGEAR-7691

Approving this PR.

aliok · 2018-07-26T07:01:16Z

@david-martin

Another thing: we need to discuss with David and the team about syncing the sync server instances. I know we talked about it and it seems necessary, but maybe it is not in the scope right now.

What do you think about this? cc @darahayes

This is about syncing the sync server states using the same mechanisms we have. @darahayes showed some great thinking and brought this, but we're not sure if this is something in scope.

Imagine the scenario:

User1 is connected to server1 (websocket connection subscriptions)
User2 is connected to server2 (websocket connection subscriptions)
User1 creates a document which should result in publishing of a subscription event
User2 won't get the event

darahayes · 2018-07-26T08:23:26Z

@aliok This will work when using the postgres pubsub implementation.

Client 1 is connected to server 1
Client 2 is connected to server 2
Client 1 performs an action (e.g. create a new resource)
Server 1 publishes this to the Postgres PubSub mechanism
Server 1 and Server 2 receive the message and notify clients as necessary.

It's basically implemented although the postgres Pubsub implementation in this PR is really a skeleton and I haven't tried it at all.

darahayes · 2018-07-26T08:26:55Z

^^ I would suggest adding the following steps to the integration testing ticket:

Get data sync server working with Postgres Pubsub implementation
Integration tests for Postgres Pubsub implementation

I feel that we should focus the integration tests much more on the Postgres implementation because the In Memory one is really not for production use.

aliok · 2018-07-26T08:32:51Z

@darahayes

This will work when using the postgres pubsub implementation.

You're right. I think I got confused with the in memory pubsub.

I would suggest adding the following steps to the integration testing ticket:

added to the ticket.

I feel that we should focus the integration tests much more on the Postgres implementation because the In Memory one is really not for production use.

We can target having Postgres pubsub for integration tests initially and can see how easy it is to create separate tests for pubsub later.

…into subscriptions

darahayes · 2018-07-26T08:54:06Z

@aliok Great! I have rebased now so ready to merge as long as you're happy!

feat: initial subscriptions implementation

a5d1073

darahayes requested a review from aliok as a code owner July 23, 2018 22:32

darahayes commented Jul 23, 2018

View reviewed changes

fix: make photoUrl optional in memeolist subscription

efee68e

darahayes requested a review from wtrocki July 23, 2018 23:56

wtrocki reviewed Jul 24, 2018

View reviewed changes

feat: Use pools to reconnect Postgres in case of connection error occurs

470528f

aliok reviewed Jul 24, 2018

View reviewed changes

wtrocki reviewed Jul 24, 2018

View reviewed changes

darahayes requested a review from david-martin July 24, 2018 09:22

fix: json issue

3e3729f

david-martin approved these changes Jul 24, 2018

View reviewed changes

Dara Hayes added 3 commits July 24, 2018 16:41

fix: add abstraction layer over pubsub mechanism

35a4a63

fix: remve var declarations

8dbe154

fix: reinitialize subscription server on schema reload

83be22c

wtrocki reviewed Jul 24, 2018

View reviewed changes

fix: ensure new schema is put into scubscription server

302168a

camilamacedo86 reviewed Jul 24, 2018

View reviewed changes

aliok mentioned this pull request Jul 26, 2018

Externalize constants into a separate file #33

Closed

aliok approved these changes Jul 26, 2018

View reviewed changes

Dara Hayes added 17 commits July 26, 2018 09:41

feat: initial subscriptions implementation

38e0479

fix: json issue

cc57e50

fix: add abstraction layer over pubsub mechanism

518de50

fix: remve var declarations

4efb686

fix: reinitialize subscription server on schema reload

bbaf082

fix: ensure new schema is put into scubscription server

db005ab

fix: remove comment

ac58bff

fix: naming

3817c01

fix: refactor conditionals in resolverMapper.js

e6e28b7

fix: allow default subscribers and publishers

53b2eeb

fix: refactor and tests

367a061

fix: rename unit test command

d090a5a

fix: typo in tests

5637b85

fix: correct call to get asyncIterator

2cb5cf4

fix: minor fixes after rebase

f9b78ae

Merge branch 'subscriptions' of github.com:aerogear/data-sync-server …

caa2b93

…into subscriptions

fix: invalid publish payload in postgres schema

2be405b

darahayes changed the title ~~[WIP] feat: initial subscriptions implementation~~ feat: initial subscriptions implementation Jul 26, 2018

darahayes mentioned this pull request Jul 26, 2018

Adds preHook and postHook fields to resolver #34

Merged

wtrocki approved these changes Jul 26, 2018

View reviewed changes

darahayes merged commit c1eb87c into master Jul 26, 2018

darahayes deleted the subscriptions branch July 26, 2018 10:40

		@@ -0,0 +1,91 @@
		const _ = require('lodash')

		module.exports = function ExpressionParser () {

Conversation

darahayes commented Jul 23, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

To do

How does the subscription piece work?

Trying out Filtering

How would this look on the server side?

Why Did I do it this way?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aliok commented Jul 24, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wtrocki commented Jul 24, 2018

Uh oh!

wtrocki commented Jul 24, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

darahayes commented Jul 24, 2018

Uh oh!

aliok commented Jul 24, 2018

Uh oh!

wtrocki commented Jul 24, 2018

Uh oh!

david-martin left a comment

Choose a reason for hiding this comment

Uh oh!

darahayes commented Jul 24, 2018

Uh oh!

wtrocki Jul 24, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

camilamacedo86 Jul 24, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

camilamacedo86 Jul 25, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

darahayes commented Jul 23, 2018 •

edited

Loading

aliok commented Jul 24, 2018 •

edited

Loading

wtrocki commented Jul 24, 2018 •

edited

Loading

wtrocki Jul 24, 2018 •

edited

Loading

camilamacedo86 Jul 24, 2018 •

edited

Loading

camilamacedo86 Jul 25, 2018 •

edited

Loading

camilamacedo86 Jul 24, 2018 •

edited

Loading

aliok Jul 26, 2018 •

edited

Loading