mirror of
https://github.com/hoernschen/dendrite.git
synced 2025-07-29 12:42:46 +00:00
Remove publicroomsapi (#1176)
* Remove all of publicroomsapi * Remove references to publicroomsapi * Remove doc references to publicroomsapi
This commit is contained in:
parent
4c1e6597c0
commit
9c1f38621c
46 changed files with 160 additions and 2528 deletions
|
@ -48,13 +48,12 @@ Arguments:
|
|||
`
|
||||
|
||||
var (
|
||||
syncServerURL = flag.String("sync-api-server-url", "", "The base URL of the listening 'dendrite-sync-api-server' process. E.g. 'http://localhost:4200'")
|
||||
clientAPIURL = flag.String("client-api-server-url", "", "The base URL of the listening 'dendrite-client-api-server' process. E.g. 'http://localhost:4321'")
|
||||
mediaAPIURL = flag.String("media-api-server-url", "", "The base URL of the listening 'dendrite-media-api-server' process. E.g. 'http://localhost:7779'")
|
||||
publicRoomsAPIURL = flag.String("public-rooms-api-server-url", "", "The base URL of the listening 'dendrite-public-rooms-api-server' process. E.g. 'http://localhost:7775'")
|
||||
bindAddress = flag.String("bind-address", ":8008", "The listening port for the proxy.")
|
||||
certFile = flag.String("tls-cert", "", "The PEM formatted X509 certificate to use for TLS")
|
||||
keyFile = flag.String("tls-key", "", "The PEM private key to use for TLS")
|
||||
syncServerURL = flag.String("sync-api-server-url", "", "The base URL of the listening 'dendrite-sync-api-server' process. E.g. 'http://localhost:4200'")
|
||||
clientAPIURL = flag.String("client-api-server-url", "", "The base URL of the listening 'dendrite-client-api-server' process. E.g. 'http://localhost:4321'")
|
||||
mediaAPIURL = flag.String("media-api-server-url", "", "The base URL of the listening 'dendrite-media-api-server' process. E.g. 'http://localhost:7779'")
|
||||
bindAddress = flag.String("bind-address", ":8008", "The listening port for the proxy.")
|
||||
certFile = flag.String("tls-cert", "", "The PEM formatted X509 certificate to use for TLS")
|
||||
keyFile = flag.String("tls-key", "", "The PEM private key to use for TLS")
|
||||
)
|
||||
|
||||
func makeProxy(targetURL string) (*httputil.ReverseProxy, error) {
|
||||
|
@ -121,13 +120,6 @@ func main() {
|
|||
fmt.Fprintln(os.Stderr, "no --media-api-server-url specified.")
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
if *publicRoomsAPIURL == "" {
|
||||
flag.Usage()
|
||||
fmt.Fprintln(os.Stderr, "no --public-rooms-api-server-url specified.")
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
syncProxy, err := makeProxy(*syncServerURL)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
|
@ -140,14 +132,8 @@ func main() {
|
|||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
publicRoomsProxy, err := makeProxy(*publicRoomsAPIURL)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
http.Handle("/_matrix/client/r0/sync", syncProxy)
|
||||
http.Handle("/_matrix/client/r0/directory/list/", publicRoomsProxy)
|
||||
http.Handle("/_matrix/client/r0/publicRooms", publicRoomsProxy)
|
||||
http.Handle("/_matrix/media/v1/", mediaProxy)
|
||||
http.Handle("/", clientProxy)
|
||||
|
||||
|
@ -159,8 +145,6 @@ func main() {
|
|||
|
||||
fmt.Println("Proxying requests to:")
|
||||
fmt.Println(" /_matrix/client/r0/sync => ", *syncServerURL+"/api/_matrix/client/r0/sync")
|
||||
fmt.Println(" /_matrix/client/r0/directory/list => ", *publicRoomsAPIURL+"/_matrix/client/r0/directory/list")
|
||||
fmt.Println(" /_matrix/client/r0/publicRooms => ", *publicRoomsAPIURL+"/_matrix/media/client/r0/publicRooms")
|
||||
fmt.Println(" /_matrix/media/v1 => ", *mediaAPIURL+"/api/_matrix/media/v1")
|
||||
fmt.Println(" /* => ", *clientAPIURL+"/api/*")
|
||||
fmt.Println("Listening on ", *bindAddress)
|
||||
|
|
|
@ -39,7 +39,7 @@ func main() {
|
|||
|
||||
clientapi.AddPublicRoutes(
|
||||
base.PublicAPIMux, base.Cfg, base.KafkaProducer, deviceDB, accountDB, federation,
|
||||
rsAPI, eduInputAPI, asQuery, stateAPI, transactions.New(), fsAPI, userAPI,
|
||||
rsAPI, eduInputAPI, asQuery, stateAPI, transactions.New(), fsAPI, userAPI, nil,
|
||||
)
|
||||
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Bind.ClientAPI), string(base.Cfg.Listen.ClientAPI))
|
||||
|
|
33
cmd/dendrite-current-state-server/main.go
Normal file
33
cmd/dendrite-current-state-server/main.go
Normal file
|
@ -0,0 +1,33 @@
|
|||
// Copyright 2020 The Matrix.org Foundation C.I.C.
|
||||
//
|
||||
// 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 main
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/currentstateserver"
|
||||
"github.com/matrix-org/dendrite/internal/setup"
|
||||
)
|
||||
|
||||
func main() {
|
||||
cfg := setup.ParseFlags(false)
|
||||
base := setup.NewBaseDendrite(cfg, "CurrentStateServer", true)
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
stateAPI := currentstateserver.NewInternalAPI(cfg, base.KafkaConsumer)
|
||||
|
||||
currentstateserver.AddInternalRoutes(base.InternalAPIMux, stateAPI)
|
||||
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Bind.CurrentState), string(base.Cfg.Listen.CurrentState))
|
||||
|
||||
}
|
|
@ -29,7 +29,6 @@ import (
|
|||
p2phttp "github.com/libp2p/go-libp2p-http"
|
||||
p2pdisc "github.com/libp2p/go-libp2p/p2p/discovery"
|
||||
"github.com/matrix-org/dendrite/appservice"
|
||||
"github.com/matrix-org/dendrite/cmd/dendrite-demo-libp2p/storage"
|
||||
"github.com/matrix-org/dendrite/currentstateserver"
|
||||
"github.com/matrix-org/dendrite/eduserver"
|
||||
"github.com/matrix-org/dendrite/federationsender"
|
||||
|
@ -130,7 +129,6 @@ func main() {
|
|||
cfg.Database.ServerKey = config.DataSource(fmt.Sprintf("file:%s-serverkey.db", *instanceName))
|
||||
cfg.Database.FederationSender = config.DataSource(fmt.Sprintf("file:%s-federationsender.db", *instanceName))
|
||||
cfg.Database.AppService = config.DataSource(fmt.Sprintf("file:%s-appservice.db", *instanceName))
|
||||
cfg.Database.PublicRoomsAPI = config.DataSource(fmt.Sprintf("file:%s-publicrooms.db", *instanceName))
|
||||
cfg.Database.Naffka = config.DataSource(fmt.Sprintf("file:%s-naffka.db", *instanceName))
|
||||
cfg.Database.CurrentState = config.DataSource(fmt.Sprintf("file:%s-currentstate.db", *instanceName))
|
||||
if err = cfg.Derive(); err != nil {
|
||||
|
@ -164,10 +162,12 @@ func main() {
|
|||
&base.Base, federation, rsAPI, keyRing,
|
||||
)
|
||||
rsAPI.SetFederationSenderAPI(fsAPI)
|
||||
/* TODO:
|
||||
publicRoomsDB, err := storage.NewPublicRoomsServerDatabaseWithPubSub(string(base.Base.Cfg.Database.PublicRoomsAPI), base.LibP2PPubsub, cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to public rooms db")
|
||||
}
|
||||
*/
|
||||
stateAPI := currentstateserver.NewInternalAPI(base.Base.Cfg, base.Base.KafkaConsumer)
|
||||
|
||||
monolith := setup.Monolith{
|
||||
|
@ -187,8 +187,6 @@ func main() {
|
|||
ServerKeyAPI: serverKeyAPI,
|
||||
StateAPI: stateAPI,
|
||||
UserAPI: userAPI,
|
||||
|
||||
PublicRoomsDB: publicRoomsDB,
|
||||
}
|
||||
monolith.AddAllPublicRoutes(base.Base.PublicAPIMux)
|
||||
|
||||
|
|
|
@ -1,164 +0,0 @@
|
|||
// Copyright 2020 The Matrix.org Foundation C.I.C.
|
||||
//
|
||||
// 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 postgreswithdht
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage/postgres"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
dht "github.com/libp2p/go-libp2p-kad-dht"
|
||||
)
|
||||
|
||||
const DHTInterval = time.Second * 10
|
||||
|
||||
// PublicRoomsServerDatabase represents a public rooms server database.
|
||||
type PublicRoomsServerDatabase struct {
|
||||
dht *dht.IpfsDHT
|
||||
postgres.PublicRoomsServerDatabase
|
||||
ourRoomsContext context.Context // our current value in the DHT
|
||||
ourRoomsCancel context.CancelFunc // cancel when we want to expire our value
|
||||
foundRooms map[string]gomatrixserverlib.PublicRoom // additional rooms we have learned about from the DHT
|
||||
foundRoomsMutex sync.RWMutex // protects foundRooms
|
||||
maintenanceTimer *time.Timer //
|
||||
roomsAdvertised atomic.Value // stores int
|
||||
roomsDiscovered atomic.Value // stores int
|
||||
}
|
||||
|
||||
// NewPublicRoomsServerDatabase creates a new public rooms server database.
|
||||
func NewPublicRoomsServerDatabase(dataSourceName string, dht *dht.IpfsDHT, localServerName gomatrixserverlib.ServerName) (*PublicRoomsServerDatabase, error) {
|
||||
pg, err := postgres.NewPublicRoomsServerDatabase(dataSourceName, nil, localServerName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
provider := PublicRoomsServerDatabase{
|
||||
dht: dht,
|
||||
PublicRoomsServerDatabase: *pg,
|
||||
}
|
||||
go provider.ResetDHTMaintenance()
|
||||
provider.roomsAdvertised.Store(0)
|
||||
provider.roomsDiscovered.Store(0)
|
||||
return &provider, nil
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) GetRoomVisibility(ctx context.Context, roomID string) (bool, error) {
|
||||
return d.PublicRoomsServerDatabase.GetRoomVisibility(ctx, roomID)
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) SetRoomVisibility(ctx context.Context, visible bool, roomID string) error {
|
||||
d.ResetDHTMaintenance()
|
||||
return d.PublicRoomsServerDatabase.SetRoomVisibility(ctx, visible, roomID)
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) CountPublicRooms(ctx context.Context) (int64, error) {
|
||||
count, err := d.PublicRoomsServerDatabase.CountPublicRooms(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
d.foundRoomsMutex.RLock()
|
||||
defer d.foundRoomsMutex.RUnlock()
|
||||
return count + int64(len(d.foundRooms)), nil
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) GetPublicRooms(ctx context.Context, offset int64, limit int16, filter string) ([]gomatrixserverlib.PublicRoom, error) {
|
||||
realfilter := filter
|
||||
if realfilter == "__local__" {
|
||||
realfilter = ""
|
||||
}
|
||||
rooms, err := d.PublicRoomsServerDatabase.GetPublicRooms(ctx, offset, limit, realfilter)
|
||||
if err != nil {
|
||||
return []gomatrixserverlib.PublicRoom{}, err
|
||||
}
|
||||
if filter != "__local__" {
|
||||
d.foundRoomsMutex.RLock()
|
||||
defer d.foundRoomsMutex.RUnlock()
|
||||
for _, room := range d.foundRooms {
|
||||
rooms = append(rooms, room)
|
||||
}
|
||||
}
|
||||
return rooms, nil
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) UpdateRoomFromEvents(ctx context.Context, eventsToAdd []gomatrixserverlib.Event, eventsToRemove []gomatrixserverlib.Event) error {
|
||||
return d.PublicRoomsServerDatabase.UpdateRoomFromEvents(ctx, eventsToAdd, eventsToRemove)
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) UpdateRoomFromEvent(ctx context.Context, event gomatrixserverlib.Event) error {
|
||||
return d.PublicRoomsServerDatabase.UpdateRoomFromEvent(ctx, event)
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) ResetDHTMaintenance() {
|
||||
if d.maintenanceTimer != nil && !d.maintenanceTimer.Stop() {
|
||||
<-d.maintenanceTimer.C
|
||||
}
|
||||
d.Interval()
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) Interval() {
|
||||
if err := d.AdvertiseRoomsIntoDHT(); err != nil {
|
||||
// fmt.Println("Failed to advertise room in DHT:", err)
|
||||
}
|
||||
if err := d.FindRoomsInDHT(); err != nil {
|
||||
// fmt.Println("Failed to find rooms in DHT:", err)
|
||||
}
|
||||
fmt.Println("Found", d.roomsDiscovered.Load(), "room(s), advertised", d.roomsAdvertised.Load(), "room(s)")
|
||||
d.maintenanceTimer = time.AfterFunc(DHTInterval, d.Interval)
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) AdvertiseRoomsIntoDHT() error {
|
||||
dbCtx, dbCancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
_ = dbCancel
|
||||
ourRooms, err := d.GetPublicRooms(dbCtx, 0, 1024, "__local__")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if j, err := json.Marshal(ourRooms); err == nil {
|
||||
d.roomsAdvertised.Store(len(ourRooms))
|
||||
d.ourRoomsContext, d.ourRoomsCancel = context.WithCancel(context.Background())
|
||||
if err := d.dht.PutValue(d.ourRoomsContext, "/matrix/publicRooms", j); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) FindRoomsInDHT() error {
|
||||
d.foundRoomsMutex.Lock()
|
||||
searchCtx, searchCancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer searchCancel()
|
||||
defer d.foundRoomsMutex.Unlock()
|
||||
results, err := d.dht.GetValues(searchCtx, "/matrix/publicRooms", 1024)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
d.foundRooms = make(map[string]gomatrixserverlib.PublicRoom)
|
||||
for _, result := range results {
|
||||
var received []gomatrixserverlib.PublicRoom
|
||||
if err := json.Unmarshal(result.Val, &received); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, room := range received {
|
||||
d.foundRooms[room.RoomID] = room
|
||||
}
|
||||
}
|
||||
d.roomsDiscovered.Store(len(d.foundRooms))
|
||||
return nil
|
||||
}
|
|
@ -1,179 +0,0 @@
|
|||
// Copyright 2020 The Matrix.org Foundation C.I.C.
|
||||
//
|
||||
// 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 postgreswithpubsub
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage/postgres"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
||||
pubsub "github.com/libp2p/go-libp2p-pubsub"
|
||||
)
|
||||
|
||||
const MaintenanceInterval = time.Second * 10
|
||||
|
||||
type discoveredRoom struct {
|
||||
time time.Time
|
||||
room gomatrixserverlib.PublicRoom
|
||||
}
|
||||
|
||||
// PublicRoomsServerDatabase represents a public rooms server database.
|
||||
type PublicRoomsServerDatabase struct {
|
||||
postgres.PublicRoomsServerDatabase //
|
||||
pubsub *pubsub.PubSub //
|
||||
subscription *pubsub.Subscription //
|
||||
foundRooms map[string]discoveredRoom // additional rooms we have learned about from the DHT
|
||||
foundRoomsMutex sync.RWMutex // protects foundRooms
|
||||
maintenanceTimer *time.Timer //
|
||||
roomsAdvertised atomic.Value // stores int
|
||||
}
|
||||
|
||||
// NewPublicRoomsServerDatabase creates a new public rooms server database.
|
||||
func NewPublicRoomsServerDatabase(dataSourceName string, pubsub *pubsub.PubSub, localServerName gomatrixserverlib.ServerName) (*PublicRoomsServerDatabase, error) {
|
||||
pg, err := postgres.NewPublicRoomsServerDatabase(dataSourceName, nil, localServerName)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
provider := PublicRoomsServerDatabase{
|
||||
pubsub: pubsub,
|
||||
PublicRoomsServerDatabase: *pg,
|
||||
foundRooms: make(map[string]discoveredRoom),
|
||||
}
|
||||
if topic, err := pubsub.Join("/matrix/publicRooms"); err != nil {
|
||||
return nil, err
|
||||
} else if sub, err := topic.Subscribe(); err == nil {
|
||||
provider.subscription = sub
|
||||
go provider.MaintenanceTimer()
|
||||
go provider.FindRooms()
|
||||
provider.roomsAdvertised.Store(0)
|
||||
return &provider, nil
|
||||
} else {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) GetRoomVisibility(ctx context.Context, roomID string) (bool, error) {
|
||||
return d.PublicRoomsServerDatabase.GetRoomVisibility(ctx, roomID)
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) SetRoomVisibility(ctx context.Context, visible bool, roomID string) error {
|
||||
d.MaintenanceTimer()
|
||||
return d.PublicRoomsServerDatabase.SetRoomVisibility(ctx, visible, roomID)
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) CountPublicRooms(ctx context.Context) (int64, error) {
|
||||
d.foundRoomsMutex.RLock()
|
||||
defer d.foundRoomsMutex.RUnlock()
|
||||
return int64(len(d.foundRooms)), nil
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) GetPublicRooms(ctx context.Context, offset int64, limit int16, filter string) ([]gomatrixserverlib.PublicRoom, error) {
|
||||
var rooms []gomatrixserverlib.PublicRoom
|
||||
if filter == "__local__" {
|
||||
if r, err := d.PublicRoomsServerDatabase.GetPublicRooms(ctx, offset, limit, ""); err == nil {
|
||||
rooms = append(rooms, r...)
|
||||
} else {
|
||||
return []gomatrixserverlib.PublicRoom{}, err
|
||||
}
|
||||
} else {
|
||||
d.foundRoomsMutex.RLock()
|
||||
defer d.foundRoomsMutex.RUnlock()
|
||||
for _, room := range d.foundRooms {
|
||||
rooms = append(rooms, room.room)
|
||||
}
|
||||
}
|
||||
return rooms, nil
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) UpdateRoomFromEvents(ctx context.Context, eventsToAdd []gomatrixserverlib.Event, eventsToRemove []gomatrixserverlib.Event) error {
|
||||
return d.PublicRoomsServerDatabase.UpdateRoomFromEvents(ctx, eventsToAdd, eventsToRemove)
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) UpdateRoomFromEvent(ctx context.Context, event gomatrixserverlib.Event) error {
|
||||
return d.PublicRoomsServerDatabase.UpdateRoomFromEvent(ctx, event)
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) MaintenanceTimer() {
|
||||
if d.maintenanceTimer != nil && !d.maintenanceTimer.Stop() {
|
||||
<-d.maintenanceTimer.C
|
||||
}
|
||||
d.Interval()
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) Interval() {
|
||||
d.foundRoomsMutex.Lock()
|
||||
for k, v := range d.foundRooms {
|
||||
if time.Since(v.time) > time.Minute {
|
||||
delete(d.foundRooms, k)
|
||||
}
|
||||
}
|
||||
d.foundRoomsMutex.Unlock()
|
||||
if err := d.AdvertiseRooms(); err != nil {
|
||||
fmt.Println("Failed to advertise room in DHT:", err)
|
||||
}
|
||||
d.foundRoomsMutex.RLock()
|
||||
defer d.foundRoomsMutex.RUnlock()
|
||||
fmt.Println("Found", len(d.foundRooms), "room(s), advertised", d.roomsAdvertised.Load(), "room(s)")
|
||||
d.maintenanceTimer = time.AfterFunc(MaintenanceInterval, d.Interval)
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) AdvertiseRooms() error {
|
||||
dbCtx, dbCancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
_ = dbCancel
|
||||
ourRooms, err := d.GetPublicRooms(dbCtx, 0, 1024, "__local__")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
advertised := 0
|
||||
for _, room := range ourRooms {
|
||||
if j, err := json.Marshal(room); err == nil {
|
||||
if topic, err := d.pubsub.Join("/matrix/publicRooms"); err != nil {
|
||||
fmt.Println("Failed to subscribe to topic:", err)
|
||||
} else if err := topic.Publish(context.TODO(), j); err != nil {
|
||||
fmt.Println("Failed to publish public room:", err)
|
||||
} else {
|
||||
advertised++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
d.roomsAdvertised.Store(advertised)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *PublicRoomsServerDatabase) FindRooms() {
|
||||
for {
|
||||
msg, err := d.subscription.Next(context.Background())
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
received := discoveredRoom{
|
||||
time: time.Now(),
|
||||
}
|
||||
if err := json.Unmarshal(msg.Data, &received.room); err != nil {
|
||||
fmt.Println("Unmarshal error:", err)
|
||||
continue
|
||||
}
|
||||
d.foundRoomsMutex.Lock()
|
||||
d.foundRooms[received.room.RoomID] = received
|
||||
d.foundRoomsMutex.Unlock()
|
||||
}
|
||||
}
|
|
@ -1,62 +0,0 @@
|
|||
// Copyright 2020 The Matrix.org Foundation C.I.C.
|
||||
//
|
||||
// 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 storage
|
||||
|
||||
import (
|
||||
"net/url"
|
||||
|
||||
dht "github.com/libp2p/go-libp2p-kad-dht"
|
||||
pubsub "github.com/libp2p/go-libp2p-pubsub"
|
||||
"github.com/matrix-org/dendrite/cmd/dendrite-demo-libp2p/storage/postgreswithdht"
|
||||
"github.com/matrix-org/dendrite/cmd/dendrite-demo-libp2p/storage/postgreswithpubsub"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage/sqlite3"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
)
|
||||
|
||||
const schemePostgres = "postgres"
|
||||
const schemeFile = "file"
|
||||
|
||||
// NewPublicRoomsServerDatabase opens a database connection.
|
||||
func NewPublicRoomsServerDatabaseWithDHT(dataSourceName string, dht *dht.IpfsDHT, localServerName gomatrixserverlib.ServerName) (storage.Database, error) {
|
||||
uri, err := url.Parse(dataSourceName)
|
||||
if err != nil {
|
||||
return postgreswithdht.NewPublicRoomsServerDatabase(dataSourceName, dht, localServerName)
|
||||
}
|
||||
switch uri.Scheme {
|
||||
case schemePostgres:
|
||||
return postgreswithdht.NewPublicRoomsServerDatabase(dataSourceName, dht, localServerName)
|
||||
case schemeFile:
|
||||
return sqlite3.NewPublicRoomsServerDatabase(dataSourceName, localServerName)
|
||||
default:
|
||||
return postgreswithdht.NewPublicRoomsServerDatabase(dataSourceName, dht, localServerName)
|
||||
}
|
||||
}
|
||||
|
||||
// NewPublicRoomsServerDatabase opens a database connection.
|
||||
func NewPublicRoomsServerDatabaseWithPubSub(dataSourceName string, pubsub *pubsub.PubSub, localServerName gomatrixserverlib.ServerName) (storage.Database, error) {
|
||||
uri, err := url.Parse(dataSourceName)
|
||||
if err != nil {
|
||||
return postgreswithpubsub.NewPublicRoomsServerDatabase(dataSourceName, pubsub, localServerName)
|
||||
}
|
||||
switch uri.Scheme {
|
||||
case schemePostgres:
|
||||
return postgreswithpubsub.NewPublicRoomsServerDatabase(dataSourceName, pubsub, localServerName)
|
||||
case schemeFile:
|
||||
return sqlite3.NewPublicRoomsServerDatabase(dataSourceName, localServerName)
|
||||
default:
|
||||
return postgreswithpubsub.NewPublicRoomsServerDatabase(dataSourceName, pubsub, localServerName)
|
||||
}
|
||||
}
|
|
@ -34,7 +34,6 @@ import (
|
|||
"github.com/matrix-org/dendrite/internal/config"
|
||||
"github.com/matrix-org/dendrite/internal/httputil"
|
||||
"github.com/matrix-org/dendrite/internal/setup"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage"
|
||||
"github.com/matrix-org/dendrite/roomserver"
|
||||
"github.com/matrix-org/dendrite/userapi"
|
||||
"github.com/matrix-org/gomatrixserverlib"
|
||||
|
@ -74,7 +73,6 @@ func main() {
|
|||
cfg.Database.ServerKey = config.DataSource(fmt.Sprintf("file:%s-serverkey.db", *instanceName))
|
||||
cfg.Database.FederationSender = config.DataSource(fmt.Sprintf("file:%s-federationsender.db", *instanceName))
|
||||
cfg.Database.AppService = config.DataSource(fmt.Sprintf("file:%s-appservice.db", *instanceName))
|
||||
cfg.Database.PublicRoomsAPI = config.DataSource(fmt.Sprintf("file:%s-publicrooms.db", *instanceName))
|
||||
cfg.Database.CurrentState = config.DataSource(fmt.Sprintf("file:%s-currentstate.db", *instanceName))
|
||||
cfg.Database.Naffka = config.DataSource(fmt.Sprintf("file:%s-naffka.db", *instanceName))
|
||||
if err = cfg.Derive(); err != nil {
|
||||
|
@ -110,11 +108,6 @@ func main() {
|
|||
|
||||
rsComponent.SetFederationSenderAPI(fsAPI)
|
||||
|
||||
publicRoomsDB, err := storage.NewPublicRoomsServerDatabase(string(base.Cfg.Database.PublicRoomsAPI), base.Cfg.DbProperties(), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to public rooms db")
|
||||
}
|
||||
|
||||
embed.Embed(base.BaseMux, *instancePort, "Yggdrasil Demo")
|
||||
|
||||
stateAPI := currentstateserver.NewInternalAPI(base.Cfg, base.KafkaConsumer)
|
||||
|
@ -136,8 +129,6 @@ func main() {
|
|||
UserAPI: userAPI,
|
||||
StateAPI: stateAPI,
|
||||
//ServerKeyAPI: serverKeyAPI,
|
||||
|
||||
PublicRoomsDB: publicRoomsDB,
|
||||
}
|
||||
monolith.AddAllPublicRoutes(base.PublicAPIMux)
|
||||
|
||||
|
|
|
@ -27,7 +27,6 @@ import (
|
|||
"github.com/matrix-org/dendrite/internal/config"
|
||||
"github.com/matrix-org/dendrite/internal/httputil"
|
||||
"github.com/matrix-org/dendrite/internal/setup"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage"
|
||||
"github.com/matrix-org/dendrite/roomserver"
|
||||
"github.com/matrix-org/dendrite/roomserver/api"
|
||||
"github.com/matrix-org/dendrite/serverkeyapi"
|
||||
|
@ -118,11 +117,6 @@ func main() {
|
|||
// This is different to rsAPI which can be the http client which doesn't need this dependency
|
||||
rsImpl.SetFederationSenderAPI(fsAPI)
|
||||
|
||||
publicRoomsDB, err := storage.NewPublicRoomsServerDatabase(string(base.Cfg.Database.PublicRoomsAPI), base.Cfg.DbProperties(), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to public rooms db")
|
||||
}
|
||||
|
||||
stateAPI := currentstateserver.NewInternalAPI(base.Cfg, base.KafkaConsumer)
|
||||
|
||||
monolith := setup.Monolith{
|
||||
|
@ -142,8 +136,6 @@ func main() {
|
|||
ServerKeyAPI: serverKeyAPI,
|
||||
StateAPI: stateAPI,
|
||||
UserAPI: userAPI,
|
||||
|
||||
PublicRoomsDB: publicRoomsDB,
|
||||
}
|
||||
monolith.AddAllPublicRoutes(base.PublicAPIMux)
|
||||
|
||||
|
|
|
@ -1,41 +0,0 @@
|
|||
// 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 main
|
||||
|
||||
import (
|
||||
"github.com/matrix-org/dendrite/internal/setup"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
func main() {
|
||||
cfg := setup.ParseFlags(false)
|
||||
base := setup.NewBaseDendrite(cfg, "PublicRoomsAPI", true)
|
||||
defer base.Close() // nolint: errcheck
|
||||
|
||||
userAPI := base.UserAPIClient()
|
||||
|
||||
rsAPI := base.RoomserverHTTPClient()
|
||||
|
||||
publicRoomsDB, err := storage.NewPublicRoomsServerDatabase(string(base.Cfg.Database.PublicRoomsAPI), base.Cfg.DbProperties(), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to public rooms db")
|
||||
}
|
||||
publicroomsapi.AddPublicRoutes(base.PublicAPIMux, base.Cfg, base.KafkaConsumer, userAPI, publicRoomsDB, rsAPI, nil, nil)
|
||||
|
||||
base.SetupAndServeHTTP(string(base.Cfg.Bind.PublicRoomsAPI), string(base.Cfg.Listen.PublicRoomsAPI))
|
||||
|
||||
}
|
|
@ -29,7 +29,6 @@ import (
|
|||
"github.com/matrix-org/dendrite/internal/config"
|
||||
"github.com/matrix-org/dendrite/internal/httputil"
|
||||
"github.com/matrix-org/dendrite/internal/setup"
|
||||
"github.com/matrix-org/dendrite/publicroomsapi/storage"
|
||||
"github.com/matrix-org/dendrite/roomserver"
|
||||
"github.com/matrix-org/dendrite/userapi"
|
||||
go_http_js_libp2p "github.com/matrix-org/go-http-js-libp2p"
|
||||
|
@ -169,7 +168,6 @@ func main() {
|
|||
cfg.Database.FederationSender = "file:/idb/dendritejs_fedsender.db"
|
||||
cfg.Database.MediaAPI = "file:/idb/dendritejs_mediaapi.db"
|
||||
cfg.Database.Naffka = "file:/idb/dendritejs_naffka.db"
|
||||
cfg.Database.PublicRoomsAPI = "file:/idb/dendritejs_publicrooms.db"
|
||||
cfg.Database.RoomServer = "file:/idb/dendritejs_roomserver.db"
|
||||
cfg.Database.ServerKey = "file:/idb/dendritejs_serverkey.db"
|
||||
cfg.Database.SyncAPI = "file:/idb/dendritejs_syncapi.db"
|
||||
|
@ -215,11 +213,6 @@ func main() {
|
|||
rsAPI.SetFederationSenderAPI(fedSenderAPI)
|
||||
p2pPublicRoomProvider := NewLibP2PPublicRoomsProvider(node, fedSenderAPI)
|
||||
|
||||
publicRoomsDB, err := storage.NewPublicRoomsServerDatabase(string(base.Cfg.Database.PublicRoomsAPI), cfg.Matrix.ServerName)
|
||||
if err != nil {
|
||||
logrus.WithError(err).Panicf("failed to connect to public rooms db")
|
||||
}
|
||||
|
||||
stateAPI := currentstateserver.NewInternalAPI(base.Cfg, base.KafkaConsumer)
|
||||
|
||||
monolith := setup.Monolith{
|
||||
|
@ -239,8 +232,6 @@ func main() {
|
|||
StateAPI: stateAPI,
|
||||
UserAPI: userAPI,
|
||||
//ServerKeyAPI: serverKeyAPI,
|
||||
|
||||
PublicRoomsDB: publicRoomsDB,
|
||||
ExtPublicRoomsProvider: p2pPublicRoomProvider,
|
||||
}
|
||||
monolith.AddAllPublicRoutes(base.PublicAPIMux)
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue