Feeds service (#24)

blog/feeds/handler/crawl.go

@@ -0,0 +1,70 @@
+package handler
+
+import (
+	"context"
+	"crypto/md5"
+	"fmt"
+
+	"net/url"
+
+	"github.com/SlyMarbo/rss"
+	log "github.com/micro/micro/v3/service/logger"
+	feeds "github.com/micro/services/blog/feeds/proto"
+	posts "github.com/micro/services/blog/posts/proto"
+)
+
+func (e *Feeds) fetchAll() {
+	fs := []*feeds.Feed{}
+	err := e.feeds.List(e.feedsNameIndex.ToQuery(nil), &fs)
+	if err != nil {
+		log.Errorf("Error listing feeds: %v", err)
+		return
+	}
+	if len(fs) == 0 {
+		log.Infof("No feeds to fetch")
+		return
+	}
+	for _, feed := range fs {
+		log.Infof("Fetching address %v", feed.Address)
+		fd, err := rss.Fetch(feed.Address)
+		if err != nil {
+			log.Errorf("Error fetching address %v: %v", feed.Address, err)
+			continue
+		}
+		domain := getDomain(feed.Address)
+
+		for _, item := range fd.Items {
+			id := fmt.Sprintf("%x", md5.Sum([]byte(item.ID)))
+			err = e.entries.Save(feeds.Entry{
+				Id:      id,
+				Url:     item.Link,
+				Title:   item.Title,
+				Domain:  domain,
+				Content: item.Summary,
+				Date:    item.Date.Unix(),
+			})
+			if err != nil {
+				log.Errorf("Error saving item: %v", err)
+			}
+			// @todo make this optional
+			_, err := e.postsService.Save(context.TODO(), &posts.SaveRequest{
+				Id:        id,
+				Title:     item.Title,
+				Content:   item.Content,
+				Timestamp: item.Date.Unix(),
+				Metadata: map[string]string{
+					"domain": domain,
+					"link":   item.Link,
+				},
+			})
+			if err != nil {
+				log.Errorf("Error saving post: %v", err)
+			}
+		}
+	}
+}
+
+func getDomain(address string) string {
+	uri, _ := url.Parse(address)
+	return uri.Host
+}

blog/feeds/handler/feeds.go

@@ -0,0 +1,78 @@
+package handler
+
+import (
+	"context"
+	"time"
+
+	"github.com/micro/dev/model"
+	log "github.com/micro/micro/v3/service/logger"
+	"github.com/micro/micro/v3/service/store"
+
+	feeds "github.com/micro/services/blog/feeds/proto"
+	posts "github.com/micro/services/blog/posts/proto"
+)
+
+type Feeds struct {
+	feeds            model.Model
+	entries          model.Model
+	postsService     posts.PostsService
+	feedsIdIndex     model.Index
+	feedsNameIndex   model.Index
+	entriesDateIndex model.Index
+}
+
+func NewFeeds(postsService posts.PostsService) *Feeds {
+	idIndex := model.ByEquality("address")
+	idIndex.Order.Type = model.OrderTypeUnordered
+
+	nameIndex := model.ByEquality("name")
+	nameIndex.Unique = true
+	nameIndex.Order.Type = model.OrderTypeUnordered
+
+	dateIndex := model.ByEquality("date")
+	dateIndex.Order.Type = model.OrderTypeDesc
+
+	f := &Feeds{
+		feeds: model.New(
+			store.DefaultStore,
+			"feeds",
+			model.Indexes(nameIndex),
+			&model.ModelOptions{
+				Debug:   false,
+				IdIndex: idIndex,
+			},
+		),
+		entries: model.New(
+			store.DefaultStore,
+			"entries",
+			model.Indexes(dateIndex),
+			&model.ModelOptions{
+				Debug: false,
+			},
+		),
+		postsService:     postsService,
+		feedsIdIndex:     idIndex,
+		feedsNameIndex:   nameIndex,
+		entriesDateIndex: dateIndex,
+	}
+
+	go f.crawl()
+	return f
+}
+
+func (e *Feeds) crawl() {
+	e.fetchAll()
+	tick := time.NewTicker(1 * time.Hour)
+	for _ = range tick.C {
+		e.fetchAll()
+	}
+}
+
+func (e *Feeds) New(ctx context.Context, req *feeds.NewRequest, rsp *feeds.NewResponse) error {
+	log.Info("Received Feeds.New request")
+	e.feeds.Save(feeds.Feed{
+		Name:    req.Name,
+		Address: req.Address,
+	})
+	return nil
+}