<file_sep>import urllib.request
from xml.etree import ElementTree as ET
from textblob import TextBlob
import pandas as pd
import operator
import json


def main():
    # fetch xml
    with urllib.request.urlopen('http://www.latimes.com/local/lanow/rss2.0.xml') as url:
        data = url.read()

    # set root element of xml file
    root = ET.fromstring(data)
    channel = root[0]

    # find all the news titles, date, author, links
    given_data = []

    for item in channel.findall('item'):
        title = str(item.find('title').text)
        date = str(item.find('pubDate').text)
        author = str(item[1].text)
        link = str(item.find('link').text)

        textForAnalyze = (str(item.find('title').text) +
                          str(item.find('description').text).replace('<p>', '').replace('</p>', '').replace('...', ''))
        polarity = TextBlob(textForAnalyze).sentiment.polarity

        given_data.append((title, date, author, link, polarity))

    labels = ['title', 'date', 'author', 'link', 'polarity']
    df = pd.DataFrame.from_records(given_data, columns=labels)
    json_f = df.reset_index().to_json(orient='records')

    print(json_f)


# start process
if __name__ == '__main__':
    main()
<file_sep>#
# created by <NAME> 5/12/2017
#
import csv
import re
import json
from math import log

class NaiveBayes:
    def __init__(self):
        self.data = {}
        self.data['count'] = 0;
        self.data['classes'] = {}

    def save(self):
        with open('naive-bayes-data.json', 'w') as outfile:
            json.dump(self.data, outfile)

    def train(self, example, classifications):
        self.data['count'] += 1
        for classification in classifications:
            if not classification in self.data['classes']:
                self.data['classes'][classification] = {}
                self.data['classes'][classification]['features'] = {}
                self.data['classes'][classification]['count'] = 0
                self.data['classes'][classification]['probability'] = 0
            self.data['classes'][classification]['count'] += 1
            self.data['classes'][classification]['probability'] = self.data['classes'][classification]['count'] / \
                                                                  self.data['count']
            for feature in example:
                if not feature in self.data['classes'][classification]['features']:
                    self.data['classes'][classification]['features'][feature] = {}
                    self.data['classes'][classification]['features'][feature]['count'] = 0
                    self.data['classes'][classification]['features'][feature]['probability'] = 0

                self.data['classes'][classification]['features'][feature]['count'] += 1
                self.data['classes'][classification]['features'][feature]['probability'] = \
                    self.data['classes'][classification]['features'][feature]['count'] / \
                    self.data['classes'][classification]['count']

    def classify(self, example):
        results = {}
        for classification in self.data['classes']:
            probability = self.data['classes'][classification]['probability'];
            for feature in example:
                if feature in self.data['classes'][classification]['features']:
                    probability *= self.data['classes'][classification]['features'][feature]['probability']
                else:
                    probability *= 0
            results[classification] = probability

        return results

class BayesTextClassifier(NaiveBayes):

    def __init__(self):
        super(BayesTextClassifier, self).__init__()

    def addStopWords(self, words):
        self.stopwords.extend(words)

    def clean_text(self, text):
        return ' '.join(re.sub("(@[A-Za-z0-9]+)|([^0-9A-Za-z \t])|(\w+:\/\/\S+)", " ", text).split())

    def create_bigrams(self, array):
        array1 = array[:]
        array2 = array[:]

        array1.pop()
        array2.pop(0)

        rv = list(map(lambda x: ' '.join(list(x)),zip(array1, array2)))
        return rv


    def train(self, text, classifications):
        text = self.clean_text(text).split(' ')
        #text.extend(list(self.create_bigrams(text)))

        return super(BayesTextClassifier, self).train(text, classifications)

    def classify(self, text):
        text = self.clean_text(text).split(' ')
        #text.extend(list(self.create_bigrams(text)))

        return super(BayesTextClassifier, self).classify(text)


def get_max_value_key(d1):
    values = list(d1.values())
    keys = list(d1.keys())
    max_value_index = values.index(max(values))
    max_key = keys[max_value_index]
    return max_key

def main():
    print("😀")

    file = open('./training_sets/Sentiment Analysis Dataset.csv', 'r')
    reader = csv.DictReader(file, delimiter=',')

    training = []
    testing = []

    i = 0
    for line in reader:
        if i > 10000000000:
            break
        if i % 10 != 0:
            training.append(line)
        else:
            testing.append(line)
        i += 1
    file.close()

    bayes = BayesTextClassifier()
    for line in training:
        bayes.train(line['text'], [line['sentiment']])

    bayes.save()

    correct = 0
    total = 0
    for line in testing:
        total += 1
        if get_max_value_key(bayes.classify(line['text'])) == line['sentiment']:
            correct += 1

    percent = correct / float(total)
    print("correct: " + str(correct) + " total: " + str(total) + " percent: " + str(percent))

    return 0


if __name__ == '__main__':
    main()
<file_sep>#!/usr/bin/env python3
import signal
import sys
import json
import os
import re

directory = os.path.dirname(os.path.abspath(__file__))

import nltk

# Set the nltk data path
nltk.data.path.append(directory + "/nltk_data")

# import vader analyzer
from textblob import TextBlob


# setup signal handling
def signal_handler(signal, frame):
    print('You pressed Ctrl+C!')
    sys.exit(0)


signal.signal(signal.SIGINT, signal_handler)


def clean_tweet(tweet):
    return ' '.join(re.sub("(@[A-Za-z0-9]+)|([^0-9A-Za-z \t])|(\w+:\/\/\S+)", " ", tweet).split())


def main():
    # Go over each line of stdin and flush after each write

    data = sys.argv[len(sys.argv) - 1]
    sentimentScores = TextBlob(clean_tweet(data))

    analysis = {}

    analysis["compound"] = sentimentScores.sentiment.polarity

    sys.stdout.write(json.dumps(analysis))
    return 0


if __name__ == '__main__':
    main()
<file_sep>#!/usr/bin/env node
/**
 * Created by Zera on 3/16/17.
 */

var Rx = require('rxjs/Rx');
const cp = require('child_process');
const LRUMap = require('lru_map').LRUMap;
const execFile = cp.execFile;
const execFileSync = cp.execFileSync;
var Twitter = require('twitter');

var client = new Twitter({
    consumer_key: process.env.TWITTER_CONSUMER_KEY,
    consumer_secret: process.env.TWITTER_CONSUMER_SECRET,
    access_token_key: process.env.TWITTER_ACCESS_TOKEN_KEY,
    access_token_secret: process.env.TWITTER_ACCESS_TOKEN_SECRET
    //bearer_token: process.env.TWITTER_BEARER_TOKEN
});

// Install NLTK data
// execFileSync('python3', ['nltk_data_install.py']);

const MAX_TRACKED_PHRASES = 400;
var termMap = new LRUMap(MAX_TRACKED_PHRASES);

var trackPhrasesSubject = (new Rx.Subject()).map(term=>{
    termMap.set(term,true);
    return Array.from(termMap.keys());
});

//switchmap the observable which tracks the terms via the twitter api
var streamObservable = trackPhrasesSubject
    .debounce(()=>Rx.Observable.timer(100))
    .switchMap(
        terms =>
            Rx.Observable.fromEvent(
                client.stream('statuses/filter',
                {
                    track:terms.toString(), //Array tostring naturally inserts a ',' between strings
                    language: "en",
                    stall_warnings: true,
                    filter_level: "none"
                }),
                'data'
            )
    ).publish(); //publish method is used to ensure only one stream is created

var tweetStreamObservable = streamObservable.refCount().filter(data => data.hasOwnProperty('text'));
var warningStreamObservable = streamObservable.filter(data => data.hasOwnProperty('warning'));
var limitStreamObservable = streamObservable.filter(data => data.hasOwnProperty('limit'));



//Create execFileObservable from execFile
var execFileObservable = Rx.Observable.bindNodeCallback(cp.execFile);

//launch a python process to analyze each tweet and flatmap the returns to the same observable
var streamAnalysis = tweetStreamObservable
    .concatMap(
        data => execFileObservable(
            'python3',
            ['-W ignore', 'analyser_vader.py', JSON.stringify(data.text) ]
        ),
        (x, y, ix, iy) => { x.sentiment = JSON.parse(y[0]); return x}
        ).share();


var phrase_streams = {};
function getStreamForPhrase(phrase){
    if (!phrase_streams[phrase]){
        phrase_streams[phrase] = Rx.Observable.defer(() => {
                trackPhrasesSubject.next(phrase);
                return streamAnalysis.filter(data => data[0].text.toLowerCase().includes(phrase.toLowerCase()))
            }
        ).share();
    }
    return phrase_streams[phrase];
}


warningStreamObservable.subscribe(data => console.error("Warning: " + data["warning"]["code"]));
//limitStreamObservable.subscribe(data => console.error("Limit: " + data["limit"]["track"]));


streamAnalysis.subscribe(data => {
    var mystring = "score:"+data.sentiment["compound"]+" Text:"+data;
    //console.log('\x1b[36m%s\x1b[0m', mystring);
    console.log(data);
});

// getStreamForPhrase("trump").subscribe(data => {
//     var mystring = "score:"+data[1]["compound"]+" Text:"+data[0].text;
//     return console.log('\x1b[31m%s\x1b[0m', mystring);
// });
//
// getStreamForPhrase("nintendo").subscribe(data => {
//     var mystring = "score:"+data[1]["compound"]+" Text:"+data[0].text;
//     return console.log('\x1b[33m%s\x1b[0m:', mystring);
// });
//
// getStreamForPhrase("obama").subscribe(data => {
//     var mystring = "score:"+data[1]["compound"]+" Text:"+data[0].text;
//     return console.log('\x1b[36m%s\x1b[0m', mystring);
// });
//
// getStreamForPhrase("twitter").subscribe(data => {
//     var mystring = "score:"+data[1]["compound"]+" Text:"+data[0].text;
//     return console.log('%s', mystring);
// });

// addPhrase("nintendo");
//
// Rx.Observable.timer(20000).subscribe(
//     ()=>addPhrase("trump")
// );
//
// Rx.Observable.timer(40000).subscribe(
//     ()=>addPhrase("obamacare")
// );

trackPhrasesSubject.next("obama");
trackPhrasesSubject.next("trump");
trackPhrasesSubject.next("nintendo");
trackPhrasesSubject.next("twitter");
trackPhrasesSubject.next("cnn");
<file_sep>import {Routes, RouterModule} from "@angular/router";
import {HomeComponent} from "./home/home.component";
import {ModuleWithProviders} from "@angular/core";
import {AnalysisComponent} from "./analysis/analysis.component";
import {LiveStreamComponent} from "./live-stream/live-stream.component";
import {LaTimesComponent} from "./la-times/la-times.component";
import {TrackedPhrasesComponent} from "./tracked-phrases/tracked-phrases.component";

const appRoutes: Routes = [
  { path: '/', pathMatch: 'full', component: HomeComponent},
  { path: 'tracked-phrases', component:TrackedPhrasesComponent },
  { path: 'live-stream', component: LiveStreamComponent},
  { path: 'la-times', component: LaTimesComponent},
  { path: 'analysis/:keyword', component: AnalysisComponent},
  { path: '**', redirectTo: '' }
];

export const routing: ModuleWithProviders = RouterModule.forRoot(appRoutes);
<file_sep># NodeSMA

### Steps to get the project running
1. install [IntelliJ IDEA Ultimate](https://www.jetbrains.com/idea/)
2. install `node.js` on your local machine
3. install [Angular-Cli](https://github.com/angular/angular-cli) with `npm install -g @angular/cli@latest`
4. run `npm install` from the command line

### Program entry points

The main entry point for the web app can be found under `WebApp/src/` most of the app files are under app. Lookup [Angular-Cli](https://github.com/angular/angular-cli) for more info on generating components and modules for Angular 2

The main entry point for the node.js/express backend is `bin/www` in the project root directory. Its setup so that anything in the /api url path will be captured by express

### Configurations

Ive set up development configurations and a production configuration

*"Dev-Serv"* and *"Dev-App"* must be run together

*"Development"* is a compound configuration and will run them together but can't be debugged with the debugger

__All Configurations will launch the server on localhost:3000__
<file_sep>import {Injectable, OnDestroy} from '@angular/core';
import * as socketio from 'socket.io-client';
import {Observable, Subject} from "rxjs/Rx";


@Injectable()
export class SocketService implements OnDestroy{

  private connectionObservable:Observable<any>;
  private connections: any = {};
  private sendSubject: Subject<any>;

  constructor() {

    //construct the message subject
    this.sendSubject = new Subject();

    //create new observable on subscription
    this.connectionObservable = Observable.defer(()=>
      //create the observable
      Observable.create((observer) => {
          //connect to the socket
          var socket = socketio.connect();

          //subscribe to messages sent
          var subscription = this.sendSubject.subscribe((message)=>socket.emit(message.label,message.data));

          observer.next(socket);

          //cleanup and close the connection
          return function () {
            subscription.unsubscribe();
            socket.disconnect();
          }
        }
      )).publishReplay().refCount();

  }

  ngOnDestroy():void {
    this.sendSubject.complete();
  }

  public messages(label:string):Observable<any> {
    //returns a new observable mapped to a function
    if (!this.connections[label]){
      this.connections[label] = this.connectionObservable
        .switchMap((socket) => Observable.fromEvent(socket,label))
        .finally(()=>this.connections[label] = null)
        .share();
    }
    return this.connections[label];
  }

  public send(label:string, data: any = {}){
    this.sendSubject.next({label:label,data:data});
  }

}
<file_sep>#!/usr/bin/env python3
import os
directory = os.path.dirname(os.path.abspath(__file__))

import nltk

# Download the vader lexicon
nltk.download('vader_lexicon', directory+"/nltk_data")
nltk.download('all-corpora', directory+"/nltk_data")

#nltk.download()
<file_sep>import { Component, OnInit } from '@angular/core';
import {Router, ActivatedRoute, Params} from "@angular/router";
import {SocketService} from "../services/socket.service";
import {Observable} from "rxjs/Observable";



@Component({
  selector: 'app-analysis',
  templateUrl: './analysis.component.html',
  styleUrls: ['./analysis.component.scss']
})
export class AnalysisComponent implements OnInit {

  constructor(private route: ActivatedRoute,
              private router: Router,
              private socket: SocketService) { }

  keyword: string;
  tweets: Observable<Array<any>>;


  ngOnInit() {
    this.route.params.subscribe(
      (params: Params)=> {
        this.keyword = params["keyword"]

        this.tweets = this.socket.messages('tweet')
          .filter(tweet=>tweet.sma_keywords.indexOf(this.keyword) != -1)
          // .sample(Observable.interval(2000))
          .scan((array: Array<any>,tweet: string)=>{
            array.unshift(tweet);
            if( array.length > 500) array.pop();
            return array
          }, []);

        this.socket.messages('search_result').take(1).subscribe(data=> console.log(data));
        this.socket.send('search', this.keyword);
      }
    );

  }

}
<file_sep>import { Component, OnInit } from '@angular/core';
import { Router } from "@angular/router";
import {Observable} from "rxjs/Observable";
import {SocketService} from "../services/socket.service";

@Component({
  selector: 'app-home',
  templateUrl: './home.component.html',
  styleUrls: ['./home.component.scss']
})

export class HomeComponent implements OnInit {

  constructor(private router: Router, private socket: SocketService) { }

  tweets: Observable<Array<any>>;

  ngOnInit() {
    this.tweets = this.socket.messages('tweet').take(3).toArray();
  }

  go(keyword: string){
    this.router.navigate(['/analysis', keyword]);
  }

}
<file_sep>/**
 * Created by Zera on 3/16/17.
 */

var MongoClient = require('mongodb').MongoClient;
var co = require('co');

var database;
var init;

module.exports = function(url){

    if (url && !init) {
        init = true;

        co(function*(){
            database = yield MongoClient.connect(url);
        }).catch(function(err){
            console.log('Error: Cannot connect to MongoDB. Make sure mongod is running.');
            process.exit(1);
        });

    }

    return database;
};



<file_sep>import {Component, OnInit, OnDestroy} from '@angular/core';
import {Observable, Subscription} from "rxjs/Rx";
import {SocketService} from "../services/socket.service";


@Component({
  selector: 'app-la-times',
  templateUrl: './la-times.component.html',
  styleUrls: ['./la-times.component.scss']
})
export class LaTimesComponent implements OnInit, OnDestroy {

  timesData: Array<any>;

  barChartData: Array<any>;
  barChartLabels: Array<any>;
  chartType: any;
  barChartOptions: any;

  constructor(private socket: SocketService) { }

  ngOnInit() {
    this.chartType = 'bar';
    this.barChartOptions = {
      scaleBeginAtZero: false,
      responsive: false,
      legend: {
        display: false
      },
      title: {
        display: true,
        text: 'Sentimental Polarity of News'
      }
    };

    this.socket.messages('times_data').take(1).subscribe(tdata=>{
      let labels: Array<any> = [];
      let dataSet: Array<number> = [];

      for (let data of tdata ){
        labels.push(data.date);
        dataSet.push(data.polarity);
      }

      this.barChartLabels = labels;
      this.barChartData = [{data:dataSet, label:'Polarity'}];

      this.timesData=tdata;

    });
    this.socket.send('get_times');
  }

  ngOnDestroy(){
  }

}
<file_sep>import {AfterContentChecked, AfterContentInit, Component, OnInit} from '@angular/core';
import {SocketService} from "../services/socket.service";
import {Observable} from "rxjs/Observable";

@Component({
  selector: 'app-tracked-phrases',
  templateUrl: './tracked-phrases.component.html',
  styleUrls: ['./tracked-phrases.component.scss']
})
export class TrackedPhrasesComponent implements OnInit, AfterContentChecked {

  trackedPhrases: Observable<Array<any>>;

  constructor(private socket: SocketService) { }

  ngOnInit() {
    this.trackedPhrases = this.socket.messages('tracked_phrases').take(1);
  }

  ngAfterContentChecked(){
    this.socket.send('get_tracked_phrases');
  }

}
<file_sep>/**
 * Created by Zera on 3/23/17.
 */

const cp = require('child_process');
const Rx = require('rxjs/Rx');
const Twitter = require('twitter');
const LRUCache = require("lru-cache");
//const db = require('./database')('mongodb://localhost:27017/SMA');

//
//Twitter Api
//

//Twitter access credentials must be provided as runtime variables

//Stream client for user streams
var streamClient = new Twitter({
    consumer_key: process.env.TWITTER_CONSUMER_KEY,
    consumer_secret: process.env.TWITTER_CONSUMER_SECRET,
    access_token_key: process.env.TWITTER_ACCESS_TOKEN_KEY,
    access_token_secret: process.env.TWITTER_ACCESS_TOKEN_SECRET
});

//Search client for app search
var searchClient = new Twitter({
    consumer_key: process.env.TWITTER_CONSUMER_KEY,
    consumer_secret: process.env.TWITTER_CONSUMER_SECRET,
    bearer_token: process.env.TWITTER_BEARER_TOKEN
});

//
//LRU Map
//

const MAX_TRACKED_PHRASES = 400;

var lru_options = {
    max: MAX_TRACKED_PHRASES,
    //length: function (n, key) { return 1 },
    //dispose: function (key, n) { n.close() },
    maxAge: 1000 * 60 * 60 // 1 hour
};

var termMap = new LRUCache(lru_options);

//
//Phrase Stream
//

//subject for injecting trends manually
var trackPhrasesSubject = new Rx.Subject();

//the trending phrases in the US every 15 seconds
var trendingPhrases = Rx.Observable.timer(0, 1000 * 15)
    .switchMapTo(searchClient.get('trends/place',{id:23424977}))
    .concatMap(data=>Rx.Observable.from(data[0].trends))
    .map(trend=>trend.name);

//combined stream of phrases exported to a list from the lru-cache
var phrasesListStream = Rx.Observable
    .from([trackPhrasesSubject,trendingPhrases])
    .flatMap(x=>x,(stream, term, streamNum,termNum)=>{
        var maxAge = streamNum == 1 ? 1000 * 60 * 60 * 2 : undefined;
        termMap.set(term, true, maxAge);
        return termMap.keys();
    });

//
//Twitter stream
//

//start the twitter stream
var streamObservable = phrasesListStream
    .sample(Rx.Observable.timer(1000, 1000 * 60 * 60)) //resample every hour but start right now
    .distinctUntilChanged() //dont reset the stream if the terms havn't changed
    .switchMap(
        terms =>{
        console.log('changing stream');

            return Rx.Observable.fromEvent(
                streamClient.stream('statuses/filter',
                    {
                        track:terms.toString(), //Array tostring naturally inserts a ',' between strings
                        language: "en",
                        stall_warnings: true,
                        filter_level: "none"
                    }),
                'data'
            );}
    ).publish(); //publish to only allow one stream

var tweetStream = streamObservable.refCount().filter(tweet=> typeof tweet.text == 'string');
//var messageStream = streamObservable.filter(tweet => tweet.text == undefined).subscribe(message=>console.error(message));

//Create execFileObservable from execFile
var execFileObservable = Rx.Observable.bindNodeCallback(cp.execFile);

//launch a python process to analyze each tweet and flatmap the returns to the same observable
var streamAnalysis = tweetStream
    .map(tweet=>{
        var text = tweet.truncated == true ? tweet.extended_tweet.full_text : tweet.text
        return {
            text:text,
            user:{
                name:tweet.user.name,
                screen_name:tweet.user.screen_name
            }
        };
    })
    //Todo: figure out a way to analyse more than one at once
    .concatMap(
        data => execFileObservable(
            'python3',
            ['-W ignore', __dirname+'/scripts/analyser_textblob.py', JSON.stringify(data.text) ]
        ),
        (x, y, ix, iy) => { x.sentiment = JSON.parse(y[0]); return x}
    ).map(tweet=>{

        tweet.sma_keywords = [];
        for(var key of termMap.keys()){
            if (tweet.text.toLowerCase().includes(key.toLowerCase())){
                tweet.sma_keywords.push(key)
            }
        }
        return tweet;
        }
    ).publish();

var streamSubscription = streamAnalysis.subscribe(data=>{
    process.send({message:"tweet", data:data});
});

var dbSubscription = streamAnalysis.groupBy(tweet=>tweet.sma_keywords[0])
    .flatMap(obs=>{
        var key = obs.key;
        var total = 3;
        return obs.windowCount(total)
            .flatMap(
                obs=>
                    obs.map(
                        tweet=>
                            tweet.sentiment.compound > 0 ? 1 : 0
                    ).count(x=>x==1))
            .map(
                count=> {
                    return {phrase:key, count:count, total:total};
                }
            );

    }).subscribe(data=>console.log(data));


//
//Messaages
//

function send(message, data, id){
    process.send({message:message, data:data, id:id});
}

//handle messages to and from the process
var messages = Rx.Observable.fromEvent(process, 'message');
messages.subscribe(message=>{
    console.log('subprocess message: ' + message.message + " from: " + (message.id ? message.id : "Server") );
    switch (message.message){
        case "start_stream":
            streamAnalysis.connect();
            send('stream_started');
            break;
        case "stop_stream":
            if (!streamSubscription) break;
            streamSubscription.unsubscribe();
            break;
        case "get_tracked_phrases":
            send('tracked_phrases', termMap.keys(), message.id);
            break;
        case "get_times":
            execFileObservable(
                'python3',
                ['-W ignore', __dirname+'/scripts/latimes.py']
            ).subscribe(data=>send('times_data', JSON.parse(data[0]), message.id));
            break;
        case "search":
            Rx.Observable.fromPromise(searchClient.get('search/tweets',{
                q:message.data,
                result_type: 'recent',
                count:100,
                language: "en"
            }))
            .subscribe(data=>send("search_result", data, message.id));
            break;
        default:
            break;
    }
});<file_sep>#!/usr/bin/env python3
import signal
import sys
import json
import os
directory = os.path.dirname(os.path.abspath(__file__))

import nltk
# Set the nltk data path
nltk.data.path.append(directory+"/nltk_data")

# nk.data.find('sentiment/vader_lexicon.zip');

#import vader analyzer
from nltk.sentiment.vader import SentimentIntensityAnalyzer


# setup signal handling
def signal_handler(signal, frame):
    print('You pressed Ctrl+C!')
    sys.exit(0)

signal.signal(signal.SIGINT, signal_handler)


def main():
# Go over each line of stdin and flush after each write
    data = sys.argv[len(sys.argv)-1]

    sid = SentimentIntensityAnalyzer()
    sentimentScores = sid.polarity_scores(data)

    sys.stdout.write(json.dumps(sentimentScores))
    return 0

if __name__ == '__main__':
    main()
<file_sep>/**
 * Created by Zera on 3/15/17.
 */
const socketio = require('socket.io');
const searcher = require('./searcher');

module.exports = function(server){
    var io = socketio(server);

    searcher.messages.subscribe(message=>{
        if (message.id){
            io.to(message.id).emit(message.message, message.data);
        }else{
            io.emit(message.message, message.data);
        }
    });

    searcher.send('start_stream');

    io.on('connection', function(client){
        console.log(client.rooms);
        client.on('get_times', (data)=>{
            searcher.send("get_times", {}, client.id)
        });
        client.on('get_tracked_phrases', (data)=>{
            searcher.send("get_tracked_phrases", {}, client.id)
        });
        client.on('search', (data)=>{
            searcher.send("search", data, client.id)
        });
    });

    io.use(function(socket, next){
        next();
    });

    // io.on('connection', function(socket){
    //
    //     console.log("connected");
    // });

    return io;
};
<file_sep>import {Component, OnInit, OnDestroy} from '@angular/core';
import {SocketService} from "../services/socket.service";
import {Observable, Subscription} from "rxjs/Rx";

@Component({
  selector: 'app-live-stream',
  templateUrl: './live-stream.component.html',
  styleUrls: ['./live-stream.component.scss']
})
export class LiveStreamComponent implements OnInit, OnDestroy{

  tweets: Observable<Array<any>>;

  constructor(private socket: SocketService) {
  }

  ngOnInit() {
    this.tweets = this.socket.messages('tweet')
      .sample(Observable.interval(1000))
      .scan((array: Array<any>,tweet: string)=>{
        array.unshift(tweet);
        if( array.length > 500) array.pop();
          return array
      }, []);
  }

  ngOnDestroy(){

  }

}
<file_sep>/**
 * Created by Zera on 3/23/17.
 */

var Rx = require('rxjs/Rx');
const cp = require('child_process');
const searcher = cp.fork(`${__dirname}/searcher_subprocess.js`);

function send(message, data, id) {
    searcher.send({message:message,data:data, id:id});
};

var messageObservable = Rx.Observable.fromEvent(searcher, 'message');

exports.send = send;
exports.startStream = ()=>send("start_stream");
exports.stopStream = ()=>send("stop_stream");
exports.search = (term)=>send("search",term);
exports.getTimes = ()=>send("get_times");

exports.disconnect = function(){
    searcher.disconnect();
};

exports.messages = messageObservable;
<file_sep>import { BrowserModule } from '@angular/platform-browser';
import { NgModule } from '@angular/core';
import { FormsModule } from '@angular/forms';
import { HttpModule } from '@angular/http';
import { CollapseModule } from 'ng2-bootstrap/collapse'

import { routing } from "./app.routing";
import { AppComponent } from './app.component';
import { HomeComponent } from './home/home.component';
import { AnalysisComponent } from './analysis/analysis.component';
import { SocketService } from "./services/socket.service";
import { LiveStreamComponent } from './live-stream/live-stream.component';
import { LaTimesComponent } from './la-times/la-times.component';
import { ChartsModule } from "ng2-charts/index";
import { TrackedPhrasesComponent } from './tracked-phrases/tracked-phrases.component';


@NgModule({
  declarations: [
    AppComponent,
    HomeComponent,
    AnalysisComponent,
    LiveStreamComponent,
    LaTimesComponent,
    TrackedPhrasesComponent
  ],
  imports: [
    BrowserModule,
    FormsModule,
    routing,
    HttpModule,
    CollapseModule,
    ChartsModule
  ],
  providers: [SocketService],
  bootstrap: [AppComponent]
})

export class AppModule {
}
