In order to use the streaming API, we will need authentication credentials from Twitter's Application Management console, much in the same way we did for our Gomniauth service providers in Chapter 3, Three Ways to Implement Profile Pictures. Head over to https://apps.twitter.com and create a new app called something like SocialPoll (the names have to be unique, so you can have some fun here; the choice of name doesn't affect the code either way). When your app has been created, visit the API Keys tab and locate the Your access token section where you need to create a new access token. After a short delay, refresh the page and notice that you in fact have two sets of keys and secrets; an API key and a secret, and an access token and the corresponding secret. Following good coding practices, we are going to set these values as environment variables so that our program can have access to them without us having to hardcode them in our source files.

The keys we will use in this chapter are:

You can set the environment variables however you like, but since the app relies on them in order to work, creating a new file called setup.sh (for bash shells) or setup.bat (on Windows) is a good idea since you can check such files into your source code repository. Insert the following code in setup.sh or setup.bat by copying the appropriate values from the Twitter app page:

#!/bin/bash
export SP_TWITTER_KEY=yCwwKKnuBnUBrelyTN...
export SP_TWITTER_SECRET=6on0YRYniT1sI3f...
export SP_TWITTER_ACCESSTOKEN=2427-13677...
export SP_TWITTER_ACCESSSECRET=SpnZf336u...

Run the file with the source or call commands to have the values appropriately set, or add them to your .bashrc or C:cmdauto.cmd files to save you running them every time you open a new terminal window.

var conn net.Conn
func dial(netw, addr string) (net.Conn, error) {
  if conn != nil {
    conn.Close()
    conn = nil
  }
  netc, err := net.DialTimeout(netw, addr, 5*time.Second)
  if err != nil {
    return nil, err
  }
  conn = netc
  return netc, nil
}

var reader io.ReadCloser
func closeConn() {
  if conn != nil {
    conn.Close()
  }
  if reader != nil {
    reader.Close()
  }
}

var (
  authClient *oauth.Client
  creds *oauth.Credentials
)
func setupTwitterAuth() {
  var ts struct {
    ConsumerKey    string `env:"SP_TWITTER_KEY,required"`
    ConsumerSecret string `env:"SP_TWITTER_SECRET,required"`
    AccessToken    string `env:"SP_TWITTER_ACCESSTOKEN,required"`
    AccessSecret   string `env:"SP_TWITTER_ACCESSSECRET,required"`
  }
  if err := envdecode.Decode(&ts); err != nil {
    log.Fatalln(err)
  }
  creds = &oauth.Credentials{
    Token:  ts.AccessToken,
    Secret: ts.AccessSecret,
  }
  authClient = &oauth.Client{
    Credentials: oauth.Credentials{
      Token:  ts.ConsumerKey,
      Secret: ts.ConsumerSecret,
    },
  }
}

var (
  authSetupOnce sync.Once
  httpClient    *http.Client
)
func makeRequest(req *http.Request, params url.Values) (*http.Response, error) {
  authSetupOnce.Do(func() {
    setupTwitterAuth()
    httpClient = &http.Client{
      Transport: &http.Transport{
        Dial: dial,
      },
    }
  })
  formEnc := params.Encode()
  req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
  req.Header.Set("Content-Length", strconv.Itoa(len(formEnc)))
  req.Header.Set("Authorization", authClient.AuthorizationHeader(creds, "POST", req.URL, params))
  return httpClient.Do(req)
}

In order to load the polls, and therefore the options to search Twitter for, we need to connect to and query MongoDB. In main.go, add the two functions dialdb and closedb:

var db *mgo.Session
func dialdb() error {
  var err error
  log.Println("dialing mongodb: localhost")
  db, err = mgo.Dial("localhost")
  return err
}
func closedb() {
  db.Close()
  log.Println("closed database connection")
}

These two functions will connect to and disconnect from the locally running MongoDB instance using the mgo package, and store mgo.Session (the database connection object) in a global variable called db.

Assuming MongoDB is running and our code is able to connect, we need to load the poll objects and extract all the options from the documents, which we will then use to search Twitter. Add the following Options function to main.go:

type poll struct {
  Options []string
}
func loadOptions() ([]string, error) {
  var options []string
  iter := db.DB("ballots").C("polls").Find(nil).Iter()
  var p poll
  for iter.Next(&p) {
    options = append(options, p.Options...)
  }
  iter.Close()
  return options, iter.Err()
}

Our poll document contains more than just Options, but our program doesn't care about anything else, so there's no need for us to bloat our poll struct. We use the db variable to access the polls collection from the ballots database, and call the mgo package's fluent Find method, passing nil (meaning no filtering).

query := col.Find(q).Sort("field").Limit(10).Skip(10)

We then get an iterator by calling the Iter method, which allows us to access each poll one by one. This is a very memory-efficient way of reading the poll data, because it only ever uses a single poll object. If we were to use the All method instead, the amount of memory we'd use would depend on the number of polls we had in our database, which would be out of our control.

When we have a poll, we use the append method to build up the options slice. Of course, with millions of polls in the database, this slice too would grow large and unwieldy. For that kind of scale, we would probably run multiple twittervotes programs, each dedicated to a portion of the poll data. A simple way to do this would be to break polls into groups based on the letters the titles begin with, such as group A-N and O-Z. A somewhat more sophisticated approach would be to add a field to the poll document grouping it up in a more controlled manner, perhaps based on the stats for the other groups so that we are able to balance the load across many twittervotes instances.

Finally, we close the iterator and clean up any used memory before returning the options and any errors that occurred while iterating (by calling the Err method on the mgo.Iter object).

Now we are able to load the options and make authorized requests to the Twitter API. We are thus ready to write the code that initiates the connection, and continuously reads from the stream until either we call our closeConn method, or Twitter closes the connection for one reason or another. The structure contained in the stream is a complex one containing all kinds of information about the tweet—who made it and when, and even what links or mentions of users occur in the body (see Twitter's API documentation for more details). However, we are only interested in the tweet text itself so you need not worry about all the other noise; add the following structure to twitter.go:

type tweet struct {
  Text string
}

Using this new structure, in twitter.go, add the following readFromTwitter function that takes a send-only channel called votes; this is how this function will inform the rest of our program that it has noticed a vote on twitter:

func readFromTwitter(votes chan<- string) {
  options, err := loadOptions()
  if err != nil {
    log.Println("failed to load options:", err)
    return
  }
  u, err := url.Parse("https://stream.twitter.com/1.1/statuses/filter.json")
  if err != nil {
    log.Println("creating filter request failed:", err)
    return
  }
  query := make(url.Values)
  query.Set("track", strings.Join(options, ","))
  req, err := http.NewRequest("POST", u.String(), strings.NewReader(query.Encode()))
  if err != nil {
    log.Println("creating filter request failed:", err)
    return
  }
  resp, err := makeRequest(req, query)
  if err != nil {
    log.Println("making request failed:", err)
    return
  }
  reader := resp.Body
  decoder := json.NewDecoder(reader)
  for {
    var tweet tweet
    if err := decoder.Decode(&tweet); err != nil {
      break
    }
    for _, option := range options {
      if strings.Contains(
        strings.ToLower(tweet.Text),
        strings.ToLower(option),
      ) {
        log.Println("vote:", option)
        votes <- option
      }
    }
  }
}

In the preceding code, after loading the options from all the polls data (by calling the loadOptions function), we use url.Parse to create a url.URL object describing the appropriate endpoint on Twitter. We build a url.Values object called query, and set the options as a comma-separated list. As per the API, we make a new POST request using the encoded url.Values object as the body, and pass it to makeRequest along with the query object itself. All being well, we make a new json.Decoder from the body of the request, and keep reading inside an infinite for loop by calling the Decode method. If there is an error (probably due to the connection being closed), we simply break the loop and exit the function. If there is a tweet to read, it will be decoded into the tweet variable, which will give us access to the Text property (the 140 characters of the tweet itself). We then iterate over all possible options, and if the tweet has mentioned it, we send it on the votes channel. This technique also allows a tweet to contain many votes at the same time, something you may or may not decide to change based on the rules of the election.

Terminating the program whenever Decode returns an error doesn't provide a very robust solution. This is because the Twitter API documentation states that the connection will drop from time to time, and clients should consider this when consuming the services. And remember, we are going to terminate the connection periodically too, so we need to think about a way to reconnect once the connection is dropped.

fmt.Println(reflect.TypeOf(true).Size())
= 1

fmt.Println(reflect.TypeOf(struct{}{}).Size())
= 0

func startTwitterStream(stopchan <-chan struct{}, votes chan<- string) <-chan struct{} {
  stoppedchan := make(chan struct{}, 1)
  go func() {
    defer func() {
      stoppedchan <- struct{}{}
    }()
    for {
      select {
      case <-stopchan:
        log.Println("stopping Twitter...")
        return
      default:
        log.Println("Querying Twitter...")
        readFromTwitter(votes)
        log.Println("  (waiting)")
        time.Sleep(10 * time.Second) // wait before reconnecting
      }
    }
  }()
  return stoppedchan
}

Once our code is successfully noticing votes on Twitter and sending them down the votes channel, we need a way to publish them into an NSQ topic; after all, this is the point of the twittervotes program.

We will write a function called publishVotes that will take the votes channel, this time of type <-chan string (a receive-only channel) and publish each string that is received from it.

Once the votes channel is closed (this is how external code will tell our function to stop working), we will stop publishing and send a signal down the returned stop signal channel.

Add the publishVotes function to main.go:

func publishVotes(votes <-chan string) <-chan struct{} {
  stopchan := make(chan struct{}, 1)
  pub, _ := nsq.NewProducer("localhost:4150", nsq.NewConfig())
  go func() {
    for vote := range votes {
      pub.Publish("votes", []byte(vote)) // publish vote
    }
    log.Println("Publisher: Stopping")
    pub.Stop()
    log.Println("Publisher: Stopped")
    stopchan <- struct{}{}
  }()
  return stopchan
}

Again the first thing we do is to create the stopchan, which we later return, this time not deferring the signaling but doing it inline by sending a struct{}{} down stopchan.

We then create an NSQ producer by calling NewProducer and connecting to the default NSQ port on localhost, using a default configuration. We start a goroutine, which uses another great built-in feature of the Go language that lets us continually pull values from a channel (in our case the votes channel) just by doing a normal for…range operation on it. Whenever the channel has no values, execution will be blocked until a value comes down the line. If the votes channel is closed, the for loop will exit.

When the loop exits (after the votes channel is closed) the publisher is stopped, following which the stopchan signal is sent.

When our program is terminated, we want to do a few things before actually exiting; namely closing our connection to Twitter and stopping the NSQ publisher (which actually deregisters its interest in the queue). To achieve this, we have to override the default Ctrl + C behavior.

Add the following code inside the main function:

var stoplock sync.Mutex
stop := false
stopChan := make(chan struct{}, 1)
signalChan := make(chan os.Signal, 1)
go func() { 
  <-signalChan
  stoplock.Lock()
  stop = true
  stoplock.Unlock()
  log.Println("Stopping...")
  stopChan <- struct{}{}
  closeConn()
}()
signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)

Here we create a stop bool with an associated sync.Mutex so that we can access it from many goroutines at the same time. We then create two more signal channels, stopChan and signalChan, and use signal.Notify to ask Go to send the signal down the signalChan when someone tries to halt the program (either with the SIGINT interrupt, or the SIGTERM termination POSIX signals). The stopChan is how we indicate that we want our processes to terminate, and we pass it as an argument to startTwitterStream later.

We then run a goroutine that blocks waiting for the signal by trying to read from signalChan; this is what the <- operator does in this case (it's trying to read from the channel). Since we don't care about the type of signal, we don't bother capturing the object returned on the channel. Once a signal is received, we set stop to true, and close the connection. Only when one of the specified signals is sent will the rest of the goroutine code run, which is how we are able to perform teardown code before exiting the program.

Add the following piece of code (inside the main function) to open and defer the closing of the database connection:

if err := dialdb(); err != nil {
  log.Fatalln("failed to dial MongoDB:", err)
}
defer closedb()

Since the readFromTwitter method reloads the options from the database each time, and because we want to keep our program updated without having to restart it, we are going to introduce one final goroutine. This goroutine will simply call closeConn every minute, causing the connection to die, and cause readFromTwitter to be called over again. Insert the following code at the bottom of the main function to start all of these processes, and then wait for them to gracefully stop:

// start things
votes := make(chan string) // chan for votes
publisherStoppedChan := publishVotes(votes)
twitterStoppedChan := startTwitterStream(stopChan, votes)
go func() {
  for {
    time.Sleep(1 * time.Minute)
    closeConn()
    stoplock.Lock()
    if stop {
      stoplock.Unlock()
      break
    }
    stoplock.Unlock()
  }
}()
<-twitterStoppedChan
close(votes)
<-publisherStoppedChan

First we make the votes channel that we have been talking about throughout this section, which is a simple channel of string. Notice that it is neither a send (chan<-) or receive (<-chan) channel; in fact, making such channels makes little sense. We then call publishVotes, passing in the votes channel for it to receive from, and capturing the returned stop signal channel as publisherStoppedChan. Similarly, we call startTwitterStream passing in our stopChan from the beginning of the main function, and the votes channel for it to send to, while capturing the resulting stop signal channel as twitterStoppedChan.

We then start our refresher goroutine, which immediately enters an infinite for loop before sleeping for a minute and closing the connection via the call to closeConn. If the stop bool has been set to true (in that previous goroutine), we will break the loop and exit, otherwise we will loop around and wait another minute before closing the connection again. The use of the stoplock is important because we have two goroutines that might try to access the stop variable at the same time but we want to avoid collisions.

Once the goroutine has started, we then block on the twitterStoppedChan by attempting to read from it. When successful (which means the signal was sent on the stopChan), we close the votes channel which will cause the publisher's for…range loop to exit, and the publisher itself to stop, after which the signal will be sent on the publisherStoppedChan, which we wait for before exiting.

To make sure our program works, we need to do two things: first we need to create a poll in the database, and second, we need to peer inside the messaging queue to see if the messages are indeed being generated by twittervotes.

In a terminal, run the mongo command to open a database shell that allows us to interact with MongoDB. Then enter the following commands to add a test poll:

> use ballots
switched to db ballots
> db.polls.insert({"title":"Test poll","options":["happy","sad","fail","win"]})

The preceding commands add a new item to the polls collection in the ballots database. We are using some common words for options that are likely to be mentioned by people on Twitter so that we can observe real tweets being translated into messages. You might notice that our poll object is missing the results field; this is fine since we are dealing with unstructured data where documents do not have to adhere to a strict schema. The counter program we are going to write in the next section will add and maintain the results data for us later.

Press Ctrl + C to exit the MongoDB shell and type the following command:

nsq_tail --topic="votes" --lookupd-http-address=localhost:4161

The nsq_tail tool connects to the specified messaging queue topic and outputs any messages that it notices. This is where we will validate that our twittervotes program is sending messages.

In a separate terminal window, let's build and run the twittervotes program:

go build –o twittervotes
./twittervotes

Now switch back to the window running nsq_tail and notice that messages are indeed being generated in response to live Twitter activity.