{"action":"created","issue":{"url":"https://api.github.com/repos/gearpump/gearpump/issues/1456","labels_url":"https://api.github.com/repos/gearpump/gearpump/issues/1456/labels{/name}","comments_url":"https://api.github.com/repos/gearpump/gearpump/issues/1456/comments","events_url":"https://api.github.com/repos/gearpump/gearpump/issues/1456/events","html_url":"https://github.com/gearpump/gearpump/issues/1456","id":109397996,"number":1456,"title":"If-statement support in DAG","user":{"login":"whjiang","id":5162753,"avatar_url":"https://avatars.githubusercontent.com/u/5162753?v=3","gravatar_id":"","url":"https://api.github.com/users/whjiang","html_url":"https://github.com/whjiang","followers_url":"https://api.github.com/users/whjiang/followers","following_url":"https://api.github.com/users/whjiang/following{/other_user}","gists_url":"https://api.github.com/users/whjiang/gists{/gist_id}","starred_url":"https://api.github.com/users/whjiang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/whjiang/subscriptions","organizations_url":"https://api.github.com/users/whjiang/orgs","repos_url":"https://api.github.com/users/whjiang/repos","events_url":"https://api.github.com/users/whjiang/events{/privacy}","received_events_url":"https://api.github.com/users/whjiang/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/gearpump/gearpump/labels/design","name":"design","color":"FF9900"},{"url":"https://api.github.com/repos/gearpump/gearpump/labels/feature","name":"feature","color":"0052cc"}],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":10,"created_at":"2015-10-01T22:34:33Z","updated_at":"2015-10-22T08:14:26Z","closed_at":null,"body":"# Goal\r\n\r\nCurrently, in Gearpump, publisher will publish each message to all the subscriptions. However, there are cases that need to selectively publish to certain subscription. E.g. in fraud detection use case, a threshold will be checked to determine which route to go (a good user, a bad user or a suspicious user?). Basically, this routing was represented as an IF-statement. \r\n\r\n```\r\nif (is_from_good_user(message)) \r\n    no more check needed\r\nelse if(is_from_bad_user(message))\r\n    alert and no more check needed\r\nelse\r\n    perform additional check to decide\r\n```\r\n\r\nTo support such routing, we need selectively route at processor level. (#1343 is on task level instead of processor level.)\r\n\r\n# Solution\r\n## solution 1\r\nSolution 1 is a workaround solution. No change need from Gearpump core part. Each If-then-else statement was represented as 2 processors\r\n```\r\nupstream ~> conditionTrueFilter ~> thenClause   #filter out the false condition messages in conditionTrueFilter\r\nupstream ~> conditionFalseFilter ~> elseClause  #filter out the true condition messages in conditionFalseFilter\r\n```\r\n\r\nThe main advantage of this solution is no need to change any code at Gearpump core side.\r\n\r\nThe main disadvantages are:\r\n*  It is hard to maintain. E.g. for dynamic DAG, if such day in future we need to change the condition, we need to carefully change both nodes. Otherwise, they will be inconsistent.\r\n* Bad performance. if the two condition filters are not at the same JVM as upstream. It means we will see significant network transport which is unnecessary. \r\n*  Hard to understand. User need to learn this BKM to write if-statement.\r\n*  Bad to express on UI. From DAG structure, it is impossible to know which one is the ```then``` clause and which one is the ```else``` clause. So, user is unable to have insight of the goodness of the condition check. E.g. does the condition check succeed in most cases?\r\n\r\n## Solution 2\r\nSolution 2 is to add built-in support for if-statement. Basically, the design is:\r\n1. Allow a processor to have more than one output channels. Each channel has a name. Each channel has a default output channel named \"out\". Processor can add alias names. \r\n2. Each channel can have multiple subscribers. Thus, for dynamic DAG, user can dynamically add/remove subscriptions for certain channel. \r\n\r\nSo, it is quite easy to implement if-statment support using this solution:\r\n* A special ```IFProcessor``` is created. It has two output channels ```then``` and ```else```. UI can show the channel name on edge. Inside ```IFProcessor```, user can write code ```output(\"then\", msg)``` to output to ```then``` channel.\r\n* DSL can be expressed like this\r\n```\r\n  upstream.if(condition, thenClause, elseClause)\r\n```\r\nor\r\n```\r\n   val ifStmt = upstream.if(condition)\r\n   ifStmt.then(thenClause)\r\n   ifStmt.else(elseClause)\r\n```\r\n* Low level API can express as \r\n```\r\n  A#\"then\" ~ partioner ~> B\r\n```"},"comment":{"url":"https://api.github.com/repos/gearpump/gearpump/issues/comments/150140564","html_url":"https://github.com/gearpump/gearpump/issues/1456#issuecomment-150140564","issue_url":"https://api.github.com/repos/gearpump/gearpump/issues/1456","id":150140564,"user":{"login":"manuzhang","id":1191767,"avatar_url":"https://avatars.githubusercontent.com/u/1191767?v=3","gravatar_id":"","url":"https://api.github.com/users/manuzhang","html_url":"https://github.com/manuzhang","followers_url":"https://api.github.com/users/manuzhang/followers","following_url":"https://api.github.com/users/manuzhang/following{/other_user}","gists_url":"https://api.github.com/users/manuzhang/gists{/gist_id}","starred_url":"https://api.github.com/users/manuzhang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/manuzhang/subscriptions","organizations_url":"https://api.github.com/users/manuzhang/orgs","repos_url":"https://api.github.com/users/manuzhang/repos","events_url":"https://api.github.com/users/manuzhang/events{/privacy}","received_events_url":"https://api.github.com/users/manuzhang/received_events","type":"User","site_admin":false},"created_at":"2015-10-22T08:14:26Z","updated_at":"2015-10-22T08:14:26Z","body":"After discussion with @clockfly, we have some concerns over the channel approach.\r\n\r\nThink about the use case that a processor A wants to selectively send messages to processors B, C or both of them.\r\n\r\nAs far as I  can see, there could be two solutions. (ch = channel)\r\n\r\n#### solution 1\r\n\r\n```scala\r\nA#ch1 ~> B\r\nA#ch2 ~> C\r\n\r\nclass A extends Task {\r\n   if (condition1) {\r\n      output(ch1, message)\r\n   } else if (condition2) {\r\n      output(ch1, message)\r\n      output(ch2, message)\r\n   } else {\r\n      output(ch2, message)\r\n   }\r\n}\r\n```\r\n\r\nThe limitations are \r\n\r\n1. what if later A only wants to send message to C on condition2 ?\r\n2. what if we want to add another processor D and A sends messages to D on a new condition ?\r\n\r\n#### solution 2 \r\n\r\n```scala\r\nA#ch1 ~> B\r\nA#ch2 ~> B\r\nA#ch2 ~> C\r\nA#ch3 ~> C\r\n\r\nclass A extends Task {\r\n  if (condition1) {\r\n     output(ch1, message)\r\n  } else if (condition2) {\r\n    output(ch2, message)\r\n  } else {\r\n    output(ch3, message)\r\n  }\r\n}\r\n```\r\n\r\nThen if A only wants to send message to B on condition1,  B can unsubscribe to channel2. However, it is still suffering if we want to add more downstream processors on new conditions.\r\n\r\n#### alternative solution\r\n\r\nAn alternative solution is to extend the capability of Partitioner to have a `filter` function.  For a DAG like\r\n\r\n```scala\r\nA ~ P1 ~> B\r\nA ~ P2 ~> C\r\n``` \r\n\r\nP1 could filter out messages that B are not interested in and P2 can do the same for C. Whenever we want to add more downstream processors, we could add such a partitioner as well. Or if later B wants to receive all messages from A, we can replace P1. \r\n\r\n\r\n\r\n\r\n"}}