123 lines
2.5 KiB
Go
123 lines
2.5 KiB
Go
package main
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"log"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
func StreamMastodon(endpoint string, o *RunningInstance) {
|
|
stream_client := http.Client{}
|
|
|
|
var oauthData OAuth
|
|
var retry bool
|
|
|
|
for {
|
|
|
|
api_timeline := "https://" + endpoint + "/api/v1/streaming/public"
|
|
req, err := http.NewRequest("GET", api_timeline, nil)
|
|
if err != nil {
|
|
log.Print("Unable to create new request")
|
|
return
|
|
}
|
|
|
|
for _, extaccount := range settings.Externalaccounts {
|
|
if extaccount.Endpoint == endpoint {
|
|
get_client(endpoint, o)
|
|
|
|
err = get_client(endpoint, o)
|
|
if err != nil {
|
|
log.Fatal("Unable to register client: ", err)
|
|
}
|
|
|
|
oauthData, err = oauth_login(endpoint, o, extaccount.Username, extaccount.Password)
|
|
if err != nil {
|
|
log.Print("Unable to login: ", err)
|
|
return
|
|
}
|
|
|
|
req.Header.Add("Authorization", oauthData.Access_token)
|
|
|
|
}
|
|
}
|
|
|
|
resp, err := stream_client.Do(req)
|
|
if err != nil {
|
|
log.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 newpost ReportPost
|
|
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), &newpost)
|
|
if err != nil {
|
|
continue
|
|
log.Fatal("Unable to unmarshal with error: ", err)
|
|
}
|
|
retry = true
|
|
default:
|
|
continue
|
|
}
|
|
default:
|
|
continue
|
|
}
|
|
|
|
go check_post(newpost.Uri)
|
|
|
|
at_sign := strings.Index(newpost.Account.Acct, "@")
|
|
newinstance := newpost.Account.Acct[at_sign+1:]
|
|
|
|
if settings.Crawl == true && stringexists(newinstance, settings.Banned) == false {
|
|
ri_mutex.Lock()
|
|
o, exists := runninginstances[newinstance]
|
|
if exists == false || o.Status == KEEPALIVE {
|
|
m := RunningInstance{}
|
|
runninginstances[newinstance] = m
|
|
go StartInstance(newinstance)
|
|
}
|
|
|
|
ri_mutex.Unlock()
|
|
}
|
|
}
|
|
if retry == true {
|
|
time.Sleep(time.Minute * 30)
|
|
} else {
|
|
break
|
|
}
|
|
}
|
|
|
|
// ri_mutex.Lock()
|
|
// m = runninginstances[endpoint]
|
|
// m.LastRun = time.Now().Format(time.RFC3339)
|
|
// m.Status = STREAM_ENDED
|
|
// runninginstances[endpoint] = m
|
|
// ri_mutex.Unlock()
|
|
}
|