Polls exists almost everywhere on the internet - Twitter, Slack - and a major similarity between all of them is the results are updated in realtime. In this tutorial, I will be describing how to build a web app that shows the popularity of a keyword in realtime with the help of Pusher Channels. The data source for our application will be tweets from Twitter.

Below is a gif of the final state of the application:

Prerequisites

• Golang >=1.5
• A Pusher account
• A Twitter application.

To do this, you need to apply as a developer before you can create an application. You can find a comprehensive guide here.

Building the application

Remember that an important step to this is to make sure you have a Twitter developer account. Kindly follow this tutorial to do that.

The next step of action is to create a directory to house our application, you will need to create a directory called streaming-api. The location of this directory will depend on the version of the Go toolchain you have - If your Go toolchain is <=1.11, you need to create the directory in your $GOPATH such as $GOPATH/src/github.com/username/streaming-api. If you are making use of >=1.12, you can create the directory literally anywhere.

Once that is done, you will need to create a file called .env, this file will contain credentials to access both the Twitter streaming API and Pusher channels. Run the command below to create the file:

    $ touch .env

Once done, you will also need to paste the following contents into the newly created .env file:

    // .env
    TWITTER_CONSUMER_KEY=TWITTER_CONSUMER_KEY
    TWITTER_CONSUMER_SECRET=TWITTER_CONSUMER_SECRET
    TWITTER_ACCESS_TOKEN=TWITTER_ACCESS_TOKEN
    TWITTER_ACCESS_SECRET=TWITTER_ACCESS_SECRET
    PUSHER_APP_ID=PUSHER_APP_ID
    PUSHER_APP_KEY=PUSHER_APP_KEY
    PUSHER_APP_SECRET=PUSHER_APP_SECRET
    PUSHER_APP_CLUSTER="eu"
    PUSHER_APP_SECURE="1"

Please remember to replace the placeholders with your actual credentials.

The next step of action is to actually create the server and the integration with Pusher Channels. To do that, you need to create a new file called main.go, that can be done by executing the command below:

    $ touch main.go

You will also need to fetch some library that are needed to help build the application. Run the command below to install these libraries:

    $ go get -v github.com/dghubble/go-twitter/twitter 
    $ go get -v github.com/dghubble/oauth1 
    $ go get -v github.com/joho/godotenv
    $ go get -v github.com/pusher/pusher-http-go

In the newly created file main.go, you will need to paste the following contents:

    // streaming-api/main.go
    
    package main
    
    import (
            "encoding/json"
            "flag"
            "fmt"
            "html/template"
            "log"
            "net/http"
            "os"
            "os/signal"
            "strings"
            "sync"
            "syscall"
            "time"
    
            "github.com/dghubble/go-twitter/twitter"
            "github.com/dghubble/oauth1"
            "github.com/joho/godotenv"
            "github.com/pusher/pusher-http-go"
    )
    
    type cache struct {
            counter map[string]int64
            mu      sync.RWMutex
    }
    
    func (c *cache) Init(options ...string) {
            for _, v := range options {
                    c.counter[strings.TrimSpace(v)] = 0
            }
    }
    
    func (c *cache) All() map[string]int64 {
            c.mu.Lock()
            defer c.mu.Unlock()
    
            return c.counter
    }
    
    func (c *cache) Incr(option string) {
            c.mu.Lock()
            defer c.mu.Unlock()
    
            c.counter[strings.TrimSpace(option)]++
    }
    
    func (c *cache) Count(option string) int64 {
            c.mu.RLock()
            defer c.mu.RUnlock()
    
            val, ok := c.counter[strings.TrimSpace(option)]
            if !ok {
                    return 0
            }
    
            return val
    }
    
    func main() {
    
            options := flag.String("options", "Messi,Suarez,Trump", "What items to search for on Twitter ?")
            httpPort := flag.Int("http.port", 1500, "What port to run HTTP on ?")
            channelsPublishInterval := flag.Duration("channels.duration", 3*time.Second, "How much duration before data is published to Pusher Channels")
    
            flag.Parse()
    
            if err := godotenv.Load(); err != nil {
                    log.Fatalf("could not load .env file.. %v", err)
            }
    
            appID := os.Getenv("PUSHER_APP_ID")
            appKey := os.Getenv("PUSHER_APP_KEY")
            appSecret := os.Getenv("PUSHER_APP_SECRET")
            appCluster := os.Getenv("PUSHER_APP_CLUSTER")
            appIsSecure := os.Getenv("PUSHER_APP_SECURE")
    
            var isSecure bool
            if appIsSecure == "1" {
                    isSecure = true
            }
    
            pusherClient := &pusher.Client{
                    AppId:   appID,
                    Key:     appKey,
                    Secret:  appSecret,
                    Cluster: appCluster,
                    Secure:  isSecure,
            }
    
            config := oauth1.NewConfig(os.Getenv("TWITTER_CONSUMER_KEY"), os.Getenv("TWITTER_CONSUMER_SECRET"))
            token := oauth1.NewToken(os.Getenv("TWITTER_ACCESS_TOKEN"), os.Getenv("TWITTER_ACCESS_SECRET"))
    
            httpClient := config.Client(oauth1.NoContext, token)
    
            client := twitter.NewClient(httpClient)
    
            optionsCache := &cache{
                    mu:      sync.RWMutex{},
                    counter: make(map[string]int64),
            }
    
            splittedOptions := strings.Split(*options, ",")
    
            if n := len(splittedOptions); n < 2 {
                    log.Fatalf("There must be at least 2 options... %v ", splittedOptions)
            } else if n > 3 {
                    log.Fatalf("There cannot be more than 3 options... %v", splittedOptions)
            }
    
            optionsCache.Init(splittedOptions...)
    
            go func() {
    
                    var t *template.Template
                    var once sync.Once
    
                    http.Handle("/static/", http.StripPrefix("/static/", http.FileServer(http.Dir("."))))
    
                    http.Handle("/polls", http.HandlerFunc(poll(optionsCache)))
                    http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
    
                            once.Do(func() {
                                    tem, err := template.ParseFiles("index.html")
                                    if err != nil {
                                            log.Fatal(err)
                                    }
    
                                    t = tem.Lookup("index.html")
                            })
    
                            t.Execute(w, nil)
                    })
    
                    http.ListenAndServe(fmt.Sprintf(":%d", *httpPort), nil)
            }()
    
            go func(c *cache, client *pusher.Client) {
    
                    t := time.NewTicker(*channelsPublishInterval)
    
                    for {
                            select {
                            case <-t.C:
                                    pusherClient.Trigger("twitter-votes", "options", c.All())
                            }
                    }
    
            }(optionsCache, pusherClient)
    
            demux := twitter.NewSwitchDemux()
            demux.Tweet = func(tweet *twitter.Tweet) {
                    for _, v := range splittedOptions {
                            if strings.Contains(tweet.Text, v) {
                                    optionsCache.Incr(v)
                            }
                    }
            }
    
            fmt.Println("Starting Stream...")
    
            filterParams := &twitter.StreamFilterParams{
                    Track:         splittedOptions,
                    StallWarnings: twitter.Bool(true),
            }
    
            stream, err := client.Streams.Filter(filterParams)
            if err != nil {
                    log.Fatal(err)
            }
    
            go demux.HandleChan(stream.Messages)
    
            ch := make(chan os.Signal)
            signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
            <-ch
    
            fmt.Println("Stopping Stream...")
            stream.Stop()
    }
    
    func poll(cache *cache) func(w http.ResponseWriter, r *http.Request) {
            return func(w http.ResponseWriter, r *http.Request) {
                    json.NewEncoder(w).Encode(cache.All())
            }
    }

While a little lengthy, the above code does just three things:

• Connect to the Twitter streaming API and listen for tweets that match our options search.
• Start an HTTP server that serves an HTML page in order to display the realtime results.
• Send an updated result to Pusher Channels.

While you might be tempted to run the application, there are still a few things missing here. We need to create one more file - index.html. This file will house the frontend for our application. You will need to go ahead to create the file by running the command below:

    $ touch index.html

In the newly created index.html file, you will need to paste the following contents in it:

    // streaming-api/index.html
    <!DOCTYPE html>
    <html lang="en">
    
    <head>
        <meta charset="UTF-8">
        <meta name="viewport" content="width=device-width, initial-scale=1.0">
        <meta http-equiv="X-UA-Compatible" content="ie=edge">
        <title>Realtime voting app based on Tweets</title>
        <link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/twitter-bootstrap/4.3.1/css/bootstrap-grid.min.css">
        <link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/Chart.js/2.8.0/Chart.min.css" />
    </head>
    
    <body>
        <div class="container">
            <div class="row">
                <div class="col-md-1">
                </div>
                <div class="col-md-10">
                    <canvas id="myChart" width="400" height="400"></canvas>
                </div>
                <div class="col-md-1">
                </div>
            </div>
        </div>
        <script src="https://cdnjs.cloudflare.com/ajax/libs/Chart.js/2.8.0/Chart.min.js"></script>
        <script src="https://cdnjs.cloudflare.com/ajax/libs/axios/0.18.0/axios.min.js"></script>
    <script src="https://js.pusher.com/4.4/pusher.min.js"></script>
    <script src="static/app.js"></script>
    </body>
    </html>

We import a few Javascript libraries but perhaps the most interesting is Line 29 which reads <script src="static/app.js"></script> . Basically, what this means is we need to create yet another file called app.js. You can go ahead to do that in the root directory with the following command:

    $ touch app.js

In the newly created app.js file, paste the following content:

    // streaming-api/app.js
    
    const APP_KEY = 'PUSHER_APP_KEY';
    const APP_CLUSTER = 'PUSHER_APP_CLUSTER';
    
    var ctx = document.getElementById('myChart').getContext('2d');
    var myChart = new Chart(ctx, {
      type: 'bar',
      data: {
        labels: [],
        datasets: [
          {
            label: '# of Tweets',
            data: [],
            backgroundColor: [
              'rgba(255, 99, 132, 0.2)',
              'rgba(54, 162, 235, 0.2)',
              'rgba(255, 159, 64, 0.2)',
            ],
            borderWidth: 1,
          },
        ],
      },
      options: {
        scales: {
          yAxes: [
            {
              ticks: {
                beginAtZero: true,
              },
            },
          ],
        },
      },
    });
    
    function updateChart(data) {
      let iterationCount = 0;
    
      for (const key in data) {
        if (!myChart.data.labels.includes(key)) {
          myChart.data.labels.push(key);
        }
    
        myChart.data.datasets.forEach(dataset => {
          dataset.data[iterationCount] = data[key];
        });
    
        iterationCount++;
    
        myChart.update();
      }
    }
    
    axios
      .get('http://localhost:1500/polls', {})
      .then(res => {
        updateChart(res.data);
      })
      .catch(err => {
        console.log('Could not retrieve information from the backend');
        console.error(err);
      });
    
    const pusher = new Pusher(APP_KEY, {
      cluster: APP_CLUSTER,
    });
    
    const channel = pusher.subscribe('twitter-votes');
    
    channel.bind('options', data => {
      updateChart(data);
    });

Please remember to make use of your actual key.

With the above done, it is time to test the application. To do this, you should run the following command in the root directory of streaming-api :

    $ go run main.go

You will need to visit http://localhost:1500 to see the chart.

You can also make use of the trending topics on your Twitter if you want to. To search Twitter for other polls, you can also make use of the following command:

    $ go run main.go -options="Apple,Javascript,Trump"

Conclusion

In this tutorial, I have described how to build a realtime popularity application that uses tweets as a data source. I also showed how to integrate with the Twitter streaming API and more importantly, Pusher Channels.

As always, the code for this tutorial can be found on GitHub.