farhan/fedilogue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

writes to database, delays based on message type

Browse Source
This commit is contained in:
farhan 2020-10-25 23:44:55 +00:00
parent 305432e58a
commit f460900f41
1 changed files with 42 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
poll/engine.go
View File

@ -2,12 +2,12 @@ package main
import (
"github.com/microcosm-cc/bluemonday"
// "github.com/jackc/pgx"
"github.com/jackc/pgx/pgxpool"
"encoding/json"
"crypto/sha1"
"io/ioutil"
"net/http"
// "context"
"context"
"strings"
"time"
"fmt"
@ -19,8 +19,9 @@ import (
type PollMessage struct {
from string
status string
status int
min_id string
numposts int
}
// Parsing Unmarshal JSON type
@ -90,7 +91,7 @@ func StartInstancePoll(endpoint string, min_id string, reportPostChan chan Repor
resp, err := http.Get(api_timeline)
if err != nil {
log.Fatal(err)
pollMessageChan <- PollMessage{endpoint, "get_failure", ""}
pollMessageChan <- PollMessage{endpoint, resp.StatusCode, "" , 0}
}
body, err := ioutil.ReadAll(resp.Body)
@ -98,9 +99,10 @@ func StartInstancePoll(endpoint string, min_id string, reportPostChan chan Repor
if err != nil {
log.Fatal(err)
panic(err)
pollMessageChan <- PollMessage{endpoint, "unmarshal_crash", ""}
pollMessageChan <- PollMessage{endpoint, resp.StatusCode, "", 0}
}
numposts := 0
for _, newpost := range newposts {
posthash := sha1.New()
@ -124,9 +126,10 @@ func StartInstancePoll(endpoint string, min_id string, reportPostChan chan Repor
if newpost.Id > min_id {
min_id = newpost.Id
}
numposts = numposts + 1
}
pollMessageChan <- PollMessage{endpoint, "ok", min_id}
pollMessageChan <- PollMessage{endpoint, resp.StatusCode, min_id, numposts}
}
func StartGetPeers(endpoint string, reportInstanceChan chan ReportInstance) {
@ -166,7 +169,18 @@ func DeferRun(pollmessage PollMessage, runninginstances *[]RunningInstance, repo
}
}
time.Sleep(time.Second * 2)
delay := 2
if pollmessage.status == 200 && pollmessage.numposts <= 10 {
delay = 2
} else if pollmessage.status == 200 && pollmessage.numposts > 10 {
delay = 5
} else if pollmessage.status == 429 {
delay = 30
} else {
fmt.Println("error, status code is: ", pollmessage.status)
os.Exit(1)
}
time.Sleep(time.Second * time.Duration(delay))
go StartInstancePoll(pollmessage.from, min_id, reportPostChan, pollMessageChan)
}
@ -185,9 +199,23 @@ func NewInstance(endpoint string, runninginstances *[]RunningInstance, reportIns
// go StartGetPeers(endpoint, reportInstanceChan)
}
func writePost(reportpost ReportPost) {
func writePost(pool *pgxpool.Pool, reportpost ReportPost) {
fmt.Println("Writing post", reportpost)
// sqlWritePost := `INSERT INTO post (url, content, strippedcontent, posthash) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (posthash) DO NOTHING`
conn, err := pool.Acquire(context.Background())
if err != nil {
fmt.Println("Error acquiring connection:", err)
os.Exit(1)
}
defer conn.Release()
_, err = conn.Exec(context.Background(), "INSERT INTO posts (url, content, strippedcontent, posthash) VALUES ($1, $2, $3, $4) ON CONFLICT (posthash) DO NOTHING", reportpost.Url, reportpost.Content, reportpost.StrippedContent, reportpost.Posthash)
if err != nil {
fmt.Println("Error on channel???")
fmt.Println(err)
os.Exit(1)
}
}
func main() {
@ -201,11 +229,11 @@ func main() {
pollMessageChan := make (chan PollMessage, 100)
// Setup Database
// dbconn, err := pgx.Connect(context.Background(), "postgres://postgres@127.0.0.1/tutorial")
// if err != nil {
// fmt.Println(err)
// os.Exit(1)
// }
pool, err := pgxpool.Connect(context.Background(), "postgres://postgres@127.0.0.1/tutorial")
if err != nil {
fmt.Fprintln(os.Stderr, "Unable to connect to database:", err)
os.Exit(1)
}
l, err := net.Listen("tcp", "127.0.0.1:5555")
if err != nil {
@ -245,7 +273,7 @@ func main() {
}
go DeferRun(p, &runninginstances, reportInstanceChan, reportPostChan, pollMessageChan)
case v := <-reportPostChan: // New Post
go writePost( v)
go writePost(pool, v)
case w := <-reportInstanceChan: // Start a new instance
NewInstance(w.endpoint, &runninginstances, reportInstanceChan, reportPostChan, pollMessageChan)
}