test/intimate
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

fix: libxml2 leak

Browse Source
This commit is contained in:
eson 2020-07-22 20:00:02 +08:00
parent b9f2f5cf22
commit 6d688b8450
8 changed files with 112 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
extractor/openrec_extractor/main.go
View File

@ -16,7 +16,6 @@ import (
*/
func main() {
go func() {
http.ListenAndServe("0.0.0.0:8899", nil)
}()

46
extractor/openrec_extractor/openrec_extractor.go
View File

@ -8,7 +8,6 @@ import (
"os"
"os/signal"
"regexp"
"runtime"
"strconv"
"strings"
"sync/atomic"
@ -18,6 +17,9 @@ import (
"github.com/tidwall/gjson"
)
var estore = intimate.NewStoreExtractor()
var sstore = intimate.NewStoreSource(string(intimate.STOpenrec))
// OpenrecExtractor 提取方法
type OpenrecExtractor struct {
user *intimate.ExtractorSource
@ -36,14 +38,27 @@ func (oe *OpenrecExtractor) Execute() {
atomic.StoreInt32(&loop, 0)
}()
estore := intimate.NewStoreExtractor()
sstore := intimate.NewStoreSource(string(intimate.STOpenrec))
var lasterr error = nil
execute := func() bool {
var err error
for atomic.LoadInt32(&loop) > 0 {
// if sstore.PopCount() >= 1000 {
// if err = estore.Close(); err != nil {
// log.Println(err)
// }
// if err = sstore.Close(); err != nil {
// log.Println(err)
// }
// estore = intimate.NewStoreExtractor()
// sstore = intimate.NewStoreSource(string(intimate.STOpenrec))
runtime.GC()
time.Sleep(time.Nanosecond)
// oe.supporters.Clear()
// oe.user.Clear()
// oe.userLive.Clear()
// runtime.GC() // 主动gc
// log.Println("1000次执行, gc 重新建立sql链接")
// }
source, err := sstore.Pop(string(intimate.TTOpenrecUser), 0)
if err != nil {
@ -52,7 +67,7 @@ func (oe *OpenrecExtractor) Execute() {
lasterr = err
}
time.Sleep(time.Second * 2)
continue
return true
}
sdata := source.Ext.([]byte)
@ -65,13 +80,16 @@ func (oe *OpenrecExtractor) Execute() {
streamer.UserId = userId
streamer.Platform = string(intimate.Popenrec)
oe.user = intimate.NewExtractorSource(datamap["html_user"])
htmlUser := datamap["html_user"]
oe.user = intimate.NewExtractorSource(&htmlUser)
oe.user.CreateExtractor()
oe.userLive = intimate.NewExtractorSource(datamap["html_live"])
htmlLive := datamap["html_live"]
oe.userLive = intimate.NewExtractorSource(&htmlLive)
oe.userLive.CreateExtractor()
oe.supporters = intimate.NewExtractorSource(datamap["json_supporters"])
jsonSupporters := datamap["json_supporters"]
oe.supporters = intimate.NewExtractorSource(&jsonSupporters)
clog := &intimate.CollectLog{}
// log.Println(anchorId)
@ -101,6 +119,13 @@ func (oe *OpenrecExtractor) Execute() {
source.Operator = int32(intimate.OperatorExtractorOK)
sstore.UpdateOperator(source)
return true
}
for atomic.LoadInt32(&loop) > 0 {
if !execute() {
break
}
}
}
@ -143,6 +168,7 @@ func (oe *OpenrecExtractor) extractViewsAndLiveStreaming(clog intimate.ISet) {
if err != nil {
log.Println(err)
}
if xp.NodeIter().Next() {
views := regexp.MustCompile(`[0-9,]+`).FindString(xp.String())
views = strings.ReplaceAll(views, ",", "")

22
extractor_field.go
View File

@ -76,21 +76,31 @@ func (cl *CollectLog) Set(field string, value interface{}) {
}
type ExtractorSource struct {
source gjson.Result
source *gjson.Result
extractor *hunter.Extractor
}
func NewExtractorSource(gr gjson.Result) *ExtractorSource {
func NewExtractorSource(gr *gjson.Result) *ExtractorSource {
es := &ExtractorSource{}
es.source = gr
es.SetSource(gr)
return es
}
func (es *ExtractorSource) CreateExtractor() {
es.extractor = hunter.NewExtractor([]byte(es.source.Str))
func (es *ExtractorSource) SetSource(gr *gjson.Result) {
es.source = gr
es.extractor = nil
}
func (es *ExtractorSource) GetSource() gjson.Result {
func (es *ExtractorSource) Clear() {
es.source = nil
es.extractor = nil
}
func (es *ExtractorSource) CreateExtractor() {
es.extractor = hunter.NewExtractor([]byte(es.source.String()))
}
func (es *ExtractorSource) GetSource() *gjson.Result {
return es.source
}

3
go.mod
View File

@ -4,9 +4,10 @@ go 1.14
require (
github.com/474420502/gcurl v0.1.2
github.com/474420502/hunter v0.3.0
github.com/474420502/hunter v0.3.4
github.com/go-sql-driver/mysql v1.5.0
github.com/lestrrat-go/libxml2 v0.0.0-20200215080510-6483566f52cb
github.com/tebeka/selenium v0.9.9
github.com/tidwall/gjson v1.6.0
github.com/tidwall/pretty v1.0.1 // indirect
golang.org/x/net v0.0.0-20200707034311-ab3426394381 // indirect

4
go.sum
View File

@ -6,8 +6,8 @@ github.com/474420502/focus v0.12.0 h1:+icbmj7IEOefvTegHt5EpcHt6WFbe2miIrceUJx2Ev
github.com/474420502/focus v0.12.0/go.mod h1:d0PMjtMxFz1a9HIhwyFPkWa+JF+0LgOrEUfd8iZka6s=
github.com/474420502/gcurl v0.1.2 h1:ON9Yz3IgAdtDlFlHfkAJ3aIEBDxH0RiViPE5ST5ohKg=
github.com/474420502/gcurl v0.1.2/go.mod h1:hws5q/Ao64bXLLDnldz9VyTQUndTWc/i5DzdEazFfoM=
github.com/474420502/hunter v0.3.0 h1:0VPi1MInxjHOta3da4v0ALWK0y3/X4/6nUSLFvdbiFU=
github.com/474420502/hunter v0.3.0/go.mod h1:pe4Xr/I+2agvq339vS/OZV+EiHAWtpXQs75rioSW9oA=
github.com/474420502/hunter v0.3.4 h1:fyLAgI84jWe3IcqsISC53j1w3CXI1FERxX//Potns0M=
github.com/474420502/hunter v0.3.4/go.mod h1:pe4Xr/I+2agvq339vS/OZV+EiHAWtpXQs75rioSW9oA=
github.com/474420502/requests v1.6.0 h1:f4h4j40eT0P5whhg9LdkotD8CaKjtuDu/vz9iSUkCgY=
github.com/474420502/requests v1.6.0/go.mod h1:SLXrQ5dL9c7dkIeKNUCBAjOIt3J9KFCS2RQjWJecNwo=
github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=

26
store.go
View File

@ -37,12 +37,22 @@ type IGetSet interface {
// SourceStore 储存
type StoreSource struct {
table string
db *sql.DB
table string
db *sql.DB
popCount int
errorCount int
errorLimit int
}
func (store *StoreSource) PopCount() int {
return store.popCount
}
func (store *StoreSource) Close() error {
return store.db.Close()
}
// NewSourceStore 创建一个存储实例
func NewStoreSource(table string) *StoreSource {
db, err := sql.Open("mysql", InitConfig.Database.SourceURI)
@ -142,6 +152,7 @@ func (store *StoreSource) Pop(targetType string, operators ...int32) (*Source, e
log.Println(err)
}
}
store.popCount++
}()
s := &Source{}
@ -150,6 +161,7 @@ func (store *StoreSource) Pop(targetType string, operators ...int32) (*Source, e
if err != nil {
return nil, err
}
s.Set("LastOperator", s.Operator)
_, err = tx.Exec("update "+store.table+" set operator = ? where uid = ?", OperatorWait, s.Uid)
return s, nil
@ -164,10 +176,19 @@ const CollectLogTable string = "collect_log"
type StoreExtractor struct {
db *sql.DB
popCount int
errorCount int
errorLimit int
}
func (store *StoreExtractor) PopCount() int {
return store.popCount
}
func (store *StoreExtractor) Close() error {
return store.db.Close()
}
func (store *StoreExtractor) errorAlarm(err error) {
if err != nil {
log.Println("store error: ", err)
@ -220,6 +241,7 @@ func (store *StoreExtractor) Pop(platform string, operators ...int32) (*Streamer
log.Println(err)
}
}
store.popCount++
}()
// log.Println(selectSQL + ` limit 1 for update`)

1
tasks/twitch/twitch_task1/task_twitch.go Normal file
View File

@ -0,0 +1 @@
package main

31
tasks/twitch/twitch_task1/task_twitch_test.go Normal file
View File

@ -0,0 +1,31 @@
package main
import (
"fmt"
"testing"
"github.com/tebeka/selenium"
"github.com/tebeka/selenium/chrome"
)
func TestCase1(t *testing.T) {
caps := selenium.Capabilities{"browserName": "chrome"}
chromecaps := chrome.Capabilities{}
err := chromecaps.AddExtension("/home/eson/test/ssh-key/0.1.2_0.crx")
if err != nil {
panic(err)
}
caps.AddChrome(chromecaps)
_, err = selenium.NewChromeDriverService("/usr/bin/chromedriver", 3030)
if err != nil {
panic(err)
}
wd, err := selenium.NewRemote(caps, fmt.Sprintf("http://localhost:%d/wd/hub", 3030))
if err != nil {
panic(err)
}
err = wd.Get("https://www.twitch.tv/directory/all")
if err != nil {
panic(err)
}
}