diff options
author | ruben <code@rbn.im> | 2019-05-21 22:56:55 +0200 |
---|---|---|
committer | Brendan Abolivier <babolivier@matrix.org> | 2019-05-21 21:56:55 +0100 |
commit | 74827428bd3e11faab65f12204449c1b9469b0ae (patch) | |
tree | 0decafa542436a0667ed2d3e3cfd4df0f03de1e5 /syncapi/syncapi.go | |
parent | 4d588f7008afe5600219ac0930c2eee2de5c447b (diff) |
use go module for dependencies (#594)
Diffstat (limited to 'syncapi/syncapi.go')
-rw-r--r-- | syncapi/syncapi.go | 75 |
1 files changed, 75 insertions, 0 deletions
diff --git a/syncapi/syncapi.go b/syncapi/syncapi.go new file mode 100644 index 00000000..2db54c3c --- /dev/null +++ b/syncapi/syncapi.go @@ -0,0 +1,75 @@ +// Copyright 2017 Vector Creations Ltd +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package syncapi + +import ( + "context" + + "github.com/sirupsen/logrus" + + "github.com/matrix-org/dendrite/clientapi/auth/storage/accounts" + "github.com/matrix-org/dendrite/common/basecomponent" + "github.com/matrix-org/dendrite/roomserver/api" + + "github.com/matrix-org/dendrite/clientapi/auth/storage/devices" + "github.com/matrix-org/dendrite/syncapi/consumers" + "github.com/matrix-org/dendrite/syncapi/routing" + "github.com/matrix-org/dendrite/syncapi/storage" + "github.com/matrix-org/dendrite/syncapi/sync" + "github.com/matrix-org/dendrite/syncapi/types" +) + +// SetupSyncAPIComponent sets up and registers HTTP handlers for the SyncAPI +// component. +func SetupSyncAPIComponent( + base *basecomponent.BaseDendrite, + deviceDB *devices.Database, + accountsDB *accounts.Database, + queryAPI api.RoomserverQueryAPI, +) { + syncDB, err := storage.NewSyncServerDatabase(string(base.Cfg.Database.SyncAPI)) + if err != nil { + logrus.WithError(err).Panicf("failed to connect to sync db") + } + + pos, err := syncDB.SyncStreamPosition(context.Background()) + if err != nil { + logrus.WithError(err).Panicf("failed to get stream position") + } + + notifier := sync.NewNotifier(types.StreamPosition(pos)) + err = notifier.Load(context.Background(), syncDB) + if err != nil { + logrus.WithError(err).Panicf("failed to start notifier") + } + + requestPool := sync.NewRequestPool(syncDB, notifier, accountsDB) + + roomConsumer := consumers.NewOutputRoomEventConsumer( + base.Cfg, base.KafkaConsumer, notifier, syncDB, queryAPI, + ) + if err = roomConsumer.Start(); err != nil { + logrus.WithError(err).Panicf("failed to start room server consumer") + } + + clientConsumer := consumers.NewOutputClientDataConsumer( + base.Cfg, base.KafkaConsumer, notifier, syncDB, + ) + if err = clientConsumer.Start(); err != nil { + logrus.WithError(err).Panicf("failed to start client data consumer") + } + + routing.Setup(base.APIMux, requestPool, syncDB, deviceDB) +} |