drone/internal/webhook/webhook.go
Johannes Batzill 7abcacdd2f feat: Add Events Framework (#120)
This Commit adds:
- stream package (provides different implementation of stream producers and consumers)
  + Redis -> will be used for any non-local deployments
  + InMemory -> a VERY BASIC implementation that is used for local execution
- events package
  + GenericReporter -> responsible for reporting events, can be used to send any type of event and payload
  + GenericReader -> responsible for reading events from a stream, can be used to register handlers for any type of event and payload
  + ReaderFactory -> responsible for launching readers for any type of consumer group&name.
- webhook package
  + The wire frame of the webhook package.
- gitrpc/events package
  + defines event Reader/Reporter for events of category git
2022-12-16 13:37:08 -08:00

52 lines
1.3 KiB
Go

// Copyright 2022 Harness Inc. All rights reserved.
// Use of this source code is governed by the Polyform Free Trial License
// that can be found in the LICENSE.md file for this repository.
package webhook
import (
"context"
"fmt"
"time"
"github.com/harness/gitness/events"
gitevents "github.com/harness/gitness/gitrpc/events"
)
const (
eventsReaderGroupName = "webhook"
processingTimeout = 2 * time.Minute
)
// Server is responsible for processing webhook events.
type Server struct {
readerCanceler *events.ReaderCanceler
}
func NewServer(ctx context.Context, gitReaderFactory *events.ReaderFactory[*gitevents.Reader],
eventsReaderName string, concurrency int) (*Server, error) {
canceler, err := gitReaderFactory.Launch(ctx, eventsReaderGroupName, eventsReaderName,
func(r *gitevents.Reader) error {
// configure reader
_ = r.SetConcurrency(concurrency)
_ = r.SetProcessingTimeout(processingTimeout)
// register events
_ = r.RegisterBranchCreated(branchCreated)
_ = r.RegisterBranchDeleted(branchDeleted)
return nil
})
if err != nil {
return nil, fmt.Errorf("failed to launch event reader for webhooks: %w", err)
}
return &Server{
readerCanceler: canceler,
}, nil
}
func (p *Server) Cancel() error {
return p.readerCanceler.Cancel()
}