Compare commits
No commits in common. "main" and "v0.1.20" have entirely different histories.
8 changed files with 10 additions and 54 deletions
|
|
@ -21,7 +21,7 @@ func main() {
|
||||||
|
|
||||||
logging.LogSetup(c.LogLevel)
|
logging.LogSetup(c.LogLevel)
|
||||||
|
|
||||||
if c.PprofEnabled {
|
if c.Metrics.GinMode != "release" {
|
||||||
go func() {
|
go func() {
|
||||||
log.Println(http.ListenAndServe("localhost:6060", nil))
|
log.Println(http.ListenAndServe("localhost:6060", nil))
|
||||||
}()
|
}()
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,6 @@ APP_LOG_LEVEL=error
|
||||||
APP_NUMCPUS=-1
|
APP_NUMCPUS=-1
|
||||||
APP_CHECK_PERIOD=6
|
APP_CHECK_PERIOD=6
|
||||||
EXTERNAL_BROWSER=
|
EXTERNAL_BROWSER=
|
||||||
PPROF_ENABLED=false
|
|
||||||
|
|
||||||
GRPC_SERVER_HOST=0.0.0.0
|
GRPC_SERVER_HOST=0.0.0.0
|
||||||
GRPC_SERVER_PORT=9060
|
GRPC_SERVER_PORT=9060
|
||||||
|
|
|
||||||
|
|
@ -7,7 +7,6 @@ import (
|
||||||
)
|
)
|
||||||
|
|
||||||
type Config struct {
|
type Config struct {
|
||||||
PprofEnabled bool
|
|
||||||
LogLevel string
|
LogLevel string
|
||||||
NumCPUs int
|
NumCPUs int
|
||||||
CheckPeriod int
|
CheckPeriod int
|
||||||
|
|
@ -47,7 +46,6 @@ type MetricsConfig struct {
|
||||||
|
|
||||||
func NewConfig() *Config {
|
func NewConfig() *Config {
|
||||||
return &Config{
|
return &Config{
|
||||||
PprofEnabled: getEnvBool("PPROF_ENABLED", true),
|
|
||||||
LogLevel: getEnv("APP_LOG_LEVEL", "debug"),
|
LogLevel: getEnv("APP_LOG_LEVEL", "debug"),
|
||||||
NumCPUs: getEnvInt("APP_NUMCPUS", -1),
|
NumCPUs: getEnvInt("APP_NUMCPUS", -1),
|
||||||
CheckPeriod: getEnvInt("APP_CHECK_PERIOD", 6),
|
CheckPeriod: getEnvInt("APP_CHECK_PERIOD", 6),
|
||||||
|
|
|
||||||
|
|
@ -141,7 +141,7 @@ func (app *App) Run(ctx context.Context) {
|
||||||
l := len(sendData)
|
l := len(sendData)
|
||||||
if l > 0 {
|
if l > 0 {
|
||||||
log.WithField("length", l).Debug("Sending parsed data")
|
log.WithField("length", l).Debug("Sending parsed data")
|
||||||
app.network.SendResult(ctx, apiClient, sendData)
|
app.network.SendResult(apiClient, sendData)
|
||||||
sendData = sendData[:0]
|
sendData = sendData[:0]
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -10,39 +10,8 @@ import (
|
||||||
"task-processor/internal/shared"
|
"task-processor/internal/shared"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (s *Parser) setupBrowser(ctx context.Context) (string, error) {
|
|
||||||
allocCtx, allocCancel := chromedp.NewRemoteAllocator(ctx, s.externalBrowser)
|
|
||||||
defer allocCancel()
|
|
||||||
|
|
||||||
pageCtx, pageCancel := chromedp.NewContext(allocCtx, chromedp.WithLogf(func(string, ...any) {}))
|
|
||||||
defer pageCancel()
|
|
||||||
|
|
||||||
copyright := "No copyright div found."
|
|
||||||
|
|
||||||
if err := chromedp.Run(pageCtx,
|
|
||||||
chromedp.Navigate("https://www.mandarake.co.jp/"),
|
|
||||||
chromedp.WaitReady("body", chromedp.ByQuery),
|
|
||||||
chromedp.Text(`div.copyright`, ©right, chromedp.ByQuery, chromedp.AtLeast(0)),
|
|
||||||
chromedp.Navigate("https://www.mandarake.co.jp/index2.html"),
|
|
||||||
chromedp.WaitReady("body", chromedp.ByQuery),
|
|
||||||
); err != nil {
|
|
||||||
log.WithError(err).Error(logHeader + logGetPrice + "failed to get single price tag")
|
|
||||||
return copyright, err
|
|
||||||
}
|
|
||||||
|
|
||||||
return copyright, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *Parser) HandleTasks(ctx context.Context, tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) {
|
func (s *Parser) HandleTasks(ctx context.Context, tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) {
|
||||||
log.Infof("%v Start handling tasks", logHeader)
|
log.Infof("%v %v handling tasks", logHeader, logWorker)
|
||||||
log.Infof("%v Setting up browser", logHeader)
|
|
||||||
cr, err := s.setupBrowser(ctx)
|
|
||||||
if err != nil {
|
|
||||||
log.WithError(err).Error(logHeader + logGetPrice + "failed to setup browser")
|
|
||||||
}
|
|
||||||
log.WithField("Copyright message", cr).Infof("%v Finished setting up browser.", logHeader)
|
|
||||||
|
|
||||||
log.Infof("%v %v processing tasks...", logHeader, logWorker)
|
|
||||||
|
|
||||||
allocCtx, allocCancel := chromedp.NewRemoteAllocator(ctx, s.externalBrowser)
|
allocCtx, allocCancel := chromedp.NewRemoteAllocator(ctx, s.externalBrowser)
|
||||||
defer allocCancel()
|
defer allocCancel()
|
||||||
|
|
|
||||||
|
|
@ -8,5 +8,5 @@ import (
|
||||||
|
|
||||||
type Handler interface {
|
type Handler interface {
|
||||||
RequestTasks(ctx context.Context, client pb.TaskProcessorClient) []shared.TaskResponse
|
RequestTasks(ctx context.Context, client pb.TaskProcessorClient) []shared.TaskResponse
|
||||||
SendResult(ctx context.Context, client pb.TaskProcessorClient, tasksDone []shared.TaskResult)
|
SendResult(client pb.TaskProcessorClient, tasksDone []shared.TaskResult)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -5,14 +5,10 @@ import (
|
||||||
log "github.com/sirupsen/logrus"
|
log "github.com/sirupsen/logrus"
|
||||||
"task-processor/internal/shared"
|
"task-processor/internal/shared"
|
||||||
pb "task-processor/proto/taskProcessor"
|
pb "task-processor/proto/taskProcessor"
|
||||||
"time"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
func (n *Network) SendResult(ctx context.Context, client pb.TaskProcessorClient, tasksDone []shared.TaskResult) {
|
func (n *Network) SendResult(client pb.TaskProcessorClient, tasksDone []shared.TaskResult) {
|
||||||
sendCtx, cancel := context.WithTimeout(ctx, time.Second*60)
|
stream, err := client.SendResult(context.Background())
|
||||||
defer cancel()
|
|
||||||
|
|
||||||
stream, err := client.SendResult(sendCtx)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatalf("Error calling PostMerch: %v", err)
|
log.Fatalf("Error calling PostMerch: %v", err)
|
||||||
}
|
}
|
||||||
|
|
@ -33,12 +29,7 @@ func (n *Network) SendResult(ctx context.Context, client pb.TaskProcessorClient,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
//if err = stream.CloseSend(); err != nil {
|
if err = stream.CloseSend(); err != nil {
|
||||||
// log.Fatalf("Error closing stream: %v", err)
|
log.Fatalf("Error closing stream: %v", err)
|
||||||
//}
|
|
||||||
|
|
||||||
_, err = stream.CloseAndRecv()
|
|
||||||
if err != nil {
|
|
||||||
log.Fatalf("Error receiving response: %v", err)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -21,7 +21,7 @@ type Deps struct {
|
||||||
const pkgLogHeader string = "Router |"
|
const pkgLogHeader string = "Router |"
|
||||||
|
|
||||||
func NewHandler(deps Deps) *Handler {
|
func NewHandler(deps Deps) *Handler {
|
||||||
engine := gin.New()
|
engine := gin.Default()
|
||||||
|
|
||||||
if deps.GinMode == "release" {
|
if deps.GinMode == "release" {
|
||||||
gin.SetMode(gin.ReleaseMode)
|
gin.SetMode(gin.ReleaseMode)
|
||||||
|
|
@ -32,8 +32,7 @@ func NewHandler(deps Deps) *Handler {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
logGroup := engine.Group("")
|
engine.GET("/", func(c *gin.Context) { c.JSON(200, gin.H{"msg": "v2"}) })
|
||||||
logGroup.GET("/", func(c *gin.Context) { c.JSON(200, gin.H{"msg": "v2"}) })
|
|
||||||
|
|
||||||
p := ginprometheus.NewPrometheus("gin")
|
p := ginprometheus.NewPrometheus("gin")
|
||||||
p.Use(engine)
|
p.Use(engine)
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue