README
¶
Event Source Atom Data: Postgres Implementation
This project processes events from the Postgres Event Store published via pgpublish to SNS and consumed by this project via an SQS queue subscribed to the topic, writing the events into organized storage to allow navigating published events using the atom protocol.
This project uses two tables: an event table to store the events associated with the atom feed, and the feed table, which keeps track of the feeds. The overall event history can be traversed from the current feed backwards using the previous entry in the feeds table. The recent items are those that have not been assigned a feedid.
As events get written to the recent table, once the size threshold for a feed is read, they are assigned a feed id. The default page size is 100 items; this may be overridden using the FEED_THRESHOLD environment variable.
Contributing
To contribute, you must certify you agree with the Developer Certificate of Origin
by signing your commits via git -s. To create a signature, configure your user name and email address in git.
Sign with your real name, do not use pseudonyms or submit anonymous commits.
In terms of workflow:
- For significant changes or improvement, create an issue before commencing work.
- Fork the respository, and create a branch for your edits.
- Add tests that cover your changes, unit tests for smaller changes, acceptance test for more significant functionality.
- Run gofmt on each file you change before committing your changes.
- Run golint on each file you change before committing your changes.
- Make sure all the tests pass before committing your changes.
- Commit your changes and issue a pull request.
License
(c) 2016 Fidelity Investments Licensed under the Apache License, Version 2.0
Documentation
¶
Index ¶
Constants ¶
const (
EnvFeedThreshold = "FEED_THRESHOLD"
)
Variables ¶
This section is empty.
Functions ¶
func RetrieveNextFeed ¶
func RetrievePreviousFeed ¶
Types ¶
type AtomDataProcessor ¶
type AtomDataProcessor struct {
// contains filtered or unexported fields
}
func NewAtomDataProcessor ¶
func NewAtomDataProcessor(db *sql.DB, env *envinject.InjectedEnv) (*AtomDataProcessor, error)
func (*AtomDataProcessor) ProcessMessage ¶
func (adp *AtomDataProcessor) ProcessMessage(msg string) error
type TimestampedEvent ¶
func RetrieveArchive ¶
func RetrieveArchive(db *sql.DB, feedid string) ([]TimestampedEvent, error)
func RetrieveEvent ¶
func RetrieveRecent ¶
func RetrieveRecent(db *sql.DB) ([]TimestampedEvent, error)
Source Files
Directories