Deletes+unboosts (#52)
* Status deletes properly streamed now. * Unboosts now work locally and federated. * Documentation updates.
This commit is contained in:
51
internal/processing/synchronous/streaming/streamdelete.go
Normal file
51
internal/processing/synchronous/streaming/streamdelete.go
Normal file
@ -0,0 +1,51 @@
|
||||
package streaming
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
|
||||
)
|
||||
|
||||
func (p *processor) StreamDelete(statusID string) error {
|
||||
errs := []string{}
|
||||
|
||||
// we want to range through ALL streams for ALL accounts here to make sure it's very clear to everyone that the status has been deleted
|
||||
p.streamMap.Range(func(k interface{}, v interface{}) bool {
|
||||
// the key of this map should be an accountID (string)
|
||||
accountID, ok := k.(string)
|
||||
if !ok {
|
||||
errs = append(errs, "key in streamMap was not a string!")
|
||||
return false
|
||||
}
|
||||
|
||||
// the value of the map should be a buncha streams
|
||||
streamsForAccount, ok := v.(*gtsmodel.StreamsForAccount)
|
||||
if !ok {
|
||||
errs = append(errs, fmt.Sprintf("stream map error for account stream %s", accountID))
|
||||
}
|
||||
|
||||
// lock the streams while we work on them
|
||||
streamsForAccount.Lock()
|
||||
defer streamsForAccount.Unlock()
|
||||
for _, stream := range streamsForAccount.Streams {
|
||||
// lock each individual stream as we work on it
|
||||
stream.Lock()
|
||||
defer stream.Unlock()
|
||||
if stream.Connected {
|
||||
stream.Messages <- >smodel.Message{
|
||||
Stream: []string{stream.Type},
|
||||
Event: "delete",
|
||||
Payload: statusID,
|
||||
}
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
|
||||
if len(errs) != 0 {
|
||||
return fmt.Errorf("one or more errors streaming status delete: %s", strings.Join(errs, ";"))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
@ -18,9 +18,14 @@ import (
|
||||
type Processor interface {
|
||||
// AuthorizeStreamingRequest returns an oauth2 token info in response to an access token query from the streaming API
|
||||
AuthorizeStreamingRequest(accessToken string) (*gtsmodel.Account, error)
|
||||
// OpenStreamForAccount returns a new Stream for the given account, which will contain a channel for passing messages back to the caller.
|
||||
OpenStreamForAccount(account *gtsmodel.Account, streamType string) (*gtsmodel.Stream, gtserror.WithCode)
|
||||
// StreamStatusToAccount streams the given status to any open, appropriate streams belonging to the given account.
|
||||
StreamStatusToAccount(s *apimodel.Status, account *gtsmodel.Account) error
|
||||
// StreamNotificationToAccount streams the given notification to any open, appropriate streams belonging to the given account.
|
||||
StreamNotificationToAccount(n *apimodel.Notification, account *gtsmodel.Account) error
|
||||
// StreamDelete streams the delete of the given statusID to *ALL* open streams.
|
||||
StreamDelete(statusID string) error
|
||||
}
|
||||
|
||||
type processor struct {
|
||||
|
Reference in New Issue
Block a user