firehose

README

Wikimedia Enterprise Firehose (realtime) API SDK

Allows to quickly connect and start using WME firehose (realtime) API.

Getting started

1. Connect to the stream:

   fhs := firehose.NewClient()
   fhs.SetAccessToken(os.Getenv("WME_ACCESS_TOKEN"))
   
   cb := func(evt *firehose.Event) {
     link, _ := url.QueryUnescape(evt.Data.URL)
     log.Printf("name: %s, identifier: %d, url: %s, dt: %s",
       evt.Data.Name,
       evt.Data.Identifier,
       link,
       evt.ID[0].Dt.Format(time.RFC3339))
   }
   
   if err := fhs.PageUpdate(context.Background(), time.Now(), cb); err != nil {
     log.Panic(err)
   }
   

Documentation

Index

• type Client
  • func NewClient() *Client
  • func (c *Client) GetAccessToken() string
  • func (c *Client) PageDelete(ctx context.Context, since time.Time, cb func(evt *Event)) error
  • func (c *Client) PageUpdate(ctx context.Context, since time.Time, cb func(evt *Event)) error
  • func (c *Client) PageVisibility(ctx context.Context, since time.Time, cb func(evt *Event)) error
  • func (c *Client) SetAccessToken(accessToken string)
• type Connection
• type ConnectionManger
  • func NewConnectionManger() *ConnectionManger
  • func (cm *ConnectionManger) Add(con *Connection)
  • func (cm *ConnectionManger) Connect(ctx context.Context, errs chan error)
• type Event
• type EventID

Types

type Client

type Client struct {
	BaseURL    string
	HTTPClient *http.Client
	// contains filtered or unexported fields
}

Client firehose streaming client to simplify work with WME realtime API.

func NewClient

func NewClient() *Client

NewClient create new realtime client.

func (*Client) GetAccessToken

func (c *Client) GetAccessToken() string

GetAccessToken returns value of the access token.

func (*Client) PageDelete

func (c *Client) PageDelete(ctx context.Context, since time.Time, cb func(evt *Event)) error

PageDelete opens connection to page delete stream.

func (*Client) PageUpdate

func (c *Client) PageUpdate(ctx context.Context, since time.Time, cb func(evt *Event)) error

PageUpdate opens connection to page update stream.

func (*Client) PageVisibility

func (c *Client) PageVisibility(ctx context.Context, since time.Time, cb func(evt *Event)) error

PageVisibility opens connection to page visibility stream.

func (*Client) SetAccessToken

func (c *Client) SetAccessToken(accessToken string)

SetAccessToken sets access token for authentication.

type Connection

type Connection struct {
	Since   time.Time
	Stream  func(ctx context.Context, since time.Time, cb func(evt *Event)) error
	Handler func(evt *Event)
}

Connection creates a single connection to the firehose stream.

type ConnectionManger

type ConnectionManger struct {
	// contains filtered or unexported fields
}

ConnectionManager helps keep connections open for firehose (realtime), satisfies "alt least once" delivery for the event.

func NewConnectionManger

func NewConnectionManger() *ConnectionManger

NewConnectionManager creates a connection manager with an empty connections list.

func (*ConnectionManger) Add

func (cm *ConnectionManger) Add(con *Connection)

Add appends new connection to the list of connections.

func (*ConnectionManger) Connect

func (cm *ConnectionManger) Connect(ctx context.Context, errs chan error)

Connect opens connections for the list in a blocking call. Leaves concurrency to the caller. Pushes connection errors to the provided channel. If you don't want error deliveries just pass nil instead of the channel.

type Event

type Event struct {
	ID   []*EventID   `json:"id"`
	Data *schema.Page `json:"data"`
}

Event server side event structure for firehose.

type EventID

type EventID struct {
	Topic     string    `json:"topic"`
	Partition int       `json:"partition"`
	Dt        time.Time `json:"dt"`
	Timestamp int64     `json:"timestamp"`
	Offset    int       `json:"offset"`
}

EventID shows metadata for the event.