Fix parser for named destination from processor

- While parser previously correctly named type as
  source leading named destination, having source and
  processor leading to named destination wrongly marked
  processor as sink. Now handling this case by
  checking presense of sinkChannel.
- Fixes #931

Author: jvalkeal

ui/src/app/shared/services/parser.spec.ts

@@ -102,21 +102,41 @@ describe('parser:', () => {
     });
 
     it('long running apps', () => {
-        parseResult = Parser.parse('aaa, bbb', 'stream');
-        expect(parseResult.lines.length).toEqual(1);
-        line = parseResult.lines[0];
-        nodes = parseResult.lines[0].nodes;
-        expect(nodes.length).toEqual(2);
-        node = nodes[0];
-        expect(node.group).toEqual('UNKNOWN_0');
-        expect(node.type).toEqual('app');
-        expect(node.name).toEqual('aaa');
-        expectRange(node.range, 0, 0, 3, 0);
-        node = nodes[1];
-        expect(node.group).toEqual('UNKNOWN_0');
-        expect(node.type).toEqual('app');
-        expect(node.name).toEqual('bbb');
-        expectRange(node.range, 5, 0, 8, 0);
+      parseResult = Parser.parse('aaa, bbb', 'stream');
+      expect(parseResult.lines.length).toEqual(1);
+      line = parseResult.lines[0];
+      nodes = parseResult.lines[0].nodes;
+      expect(nodes.length).toEqual(2);
+      node = nodes[0];
+      expect(node.group).toEqual('UNKNOWN_0');
+      expect(node.type).toEqual('app');
+      expect(node.name).toEqual('aaa');
+      expectRange(node.range, 0, 0, 3, 0);
+      node = nodes[1];
+      expect(node.group).toEqual('UNKNOWN_0');
+      expect(node.type).toEqual('app');
+      expect(node.name).toEqual('bbb');
+      expectRange(node.range, 5, 0, 8, 0);
+    });
+
+    it('from source and processor to named destination', () => {
+      parseResult = Parser.parse('time | transform > :foo', 'stream');
+      expect(parseResult.lines.length).toEqual(1);
+      line = parseResult.lines[0];
+      nodes = parseResult.lines[0].nodes;
+      expect(nodes.length).toEqual(2);
+      node = nodes[0];
+      expect(node.group).toEqual('UNKNOWN_0');
+      expect(node.type).toEqual('source');
+      expect(node.name).toEqual('time');
+      expectRange(node.range, 0, 0, 4, 0);
+      expectChannels(node, null, null);
+      node = nodes[1];
+      expect(node.group).toEqual('UNKNOWN_0');
+      expect(node.type).toEqual('processor');
+      expect(node.name).toEqual('transform');
+      expectRange(node.range, 7, 0, 16, 0);
+      expectChannels(node, null, 'foo');
     });
 
     it('options', () => {

ui/src/app/shared/services/parser.ts

@@ -587,7 +587,10 @@ class InternalParser {
                             } else {
                                 if (m === 0 && !streamdef.sourceChannel) {
                                     expectedType = 'source';
-                                } else if (m === (streamdef.apps.length - 1)) {
+                                } else if (m === (streamdef.apps.length - 1) && !streamdef.sinkChannel) {
+                                    // if last expect it to be sink only
+                                    // without sink channel. i.e. source | processor > :dest
+                                    // we fall back to processor type
                                     expectedType = 'sink';
                                 }
                             }