118 lines
2.5 KiB
Go
118 lines
2.5 KiB
Go
package main
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
// "net"
|
|
)
|
|
|
|
func StreamMastodon(endpoint string, o *RunningInstance) {
|
|
stream_client := BuildClient(endpoint)
|
|
|
|
var oauthData OAuth
|
|
var retry bool
|
|
|
|
for {
|
|
|
|
api_timeline := "https://" + endpoint + "/api/v1/streaming/public"
|
|
req, err := http.NewRequest("GET", api_timeline, nil)
|
|
req.Header.Set("User-Agent", "Tusky")
|
|
if err != nil {
|
|
logFatal.Fatal("Unable to create new request for " + endpoint + ", exiting.")
|
|
return
|
|
}
|
|
|
|
for _, extaccount := range settings.Externalaccounts {
|
|
if extaccount.Endpoint == endpoint {
|
|
get_client(endpoint, o)
|
|
|
|
err = get_client(endpoint, o)
|
|
if err != nil {
|
|
logWarn.Print("Unable to register client: ", err)
|
|
}
|
|
|
|
oauthData, err = oauth_login(endpoint, o, extaccount.Username, extaccount.Password)
|
|
if err != nil {
|
|
logWarn.Print("Unable to login: ", err)
|
|
return
|
|
}
|
|
|
|
req.Header.Add("Authorization", oauthData.Access_token)
|
|
|
|
}
|
|
}
|
|
|
|
var resp *http.Response
|
|
|
|
for tries := 0; tries < 10; tries++ {
|
|
resp, err = stream_client.Do(req)
|
|
if err != nil {
|
|
time.Sleep(time.Minute * 5)
|
|
logWarn.Print("Failure connecting to " + req.URL.Scheme + "://" + req.URL.Host + req.URL.Path + ", attempt ", tries + 1, ", sleeping for 5 minutes, ", err)
|
|
continue
|
|
}
|
|
break
|
|
}
|
|
if err != nil {
|
|
logErr.Print("Unable to stream " + api_timeline + ": ", err)
|
|
return
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
ri_mutex.Lock()
|
|
m := runninginstances[endpoint]
|
|
m.Status = RUNNING
|
|
m.LastRun = "Streaming"
|
|
runninginstances[endpoint] = m
|
|
ri_mutex.Unlock()
|
|
|
|
s := bufio.NewScanner(resp.Body)
|
|
var name string
|
|
for s.Scan() {
|
|
line := s.Text()
|
|
token := strings.SplitN(line, ":", 2)
|
|
var newactivity ReportActivity
|
|
if len(token) != 2 {
|
|
continue
|
|
}
|
|
|
|
switch strings.TrimSpace(token[0]) {
|
|
case "event":
|
|
name = strings.TrimSpace(token[1])
|
|
continue
|
|
case "data":
|
|
switch name {
|
|
case "update":
|
|
jsondata := token[1][1:]
|
|
err := json.Unmarshal([]byte(jsondata), &newactivity)
|
|
if err != nil {
|
|
logDebug.Print("Unable to parse data from "+endpoint+", but still connected.")
|
|
continue
|
|
}
|
|
retry = true
|
|
default:
|
|
continue
|
|
}
|
|
default:
|
|
continue
|
|
}
|
|
|
|
go check_activity(newactivity.Uri)
|
|
|
|
matchset := re.FindStringSubmatch(newactivity.Uri)
|
|
if matchset != nil {
|
|
newinstance := matchset[1]
|
|
go CheckInstance(newinstance, endpoint)
|
|
}
|
|
}
|
|
if retry == true {
|
|
time.Sleep(time.Minute * 30)
|
|
} else {
|
|
break
|
|
}
|
|
}
|
|
}
|