mirror of
https://github.com/superseriousbusiness/gotosocial.git
synced 2024-12-04 22:20:53 +01:00
7ec1e1332e
* rewrite cache library as codeberg.org/gruf/go-structr, implement in gotosocial
* use actual go-structr release version (not just commit hash)
* revert go toolchain changes (damn you go for auto changing this)
* fix go mod woes
* ensure %w is used in calls to errs.Appendf()
* fix error checking
* fix possible panic
* remove unnecessary start/stop functions, move to main Cache{} struct, add note regarding which caches require start/stop
* fix copy-paste artifact... 😇
* fix all comment copy-paste artifacts
* remove dropID() function, now we can just use slices.DeleteFunc()
* use util.Deduplicate() instead of collate(), move collate to util
* move orderByIDs() to util package and "generify"
* add a util.DeleteIf() function, use this to delete entries on failed population
* use slices.DeleteFunc() instead of util.DeleteIf() (i had the logic mixed up in my head somehow lol)
* add note about how collate differs from deduplicate
138 lines
3.4 KiB
Go
138 lines
3.4 KiB
Go
// GoToSocial
|
|
// Copyright (C) GoToSocial Authors admin@gotosocial.org
|
|
// SPDX-License-Identifier: AGPL-3.0-or-later
|
|
//
|
|
// This program is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU Affero General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// This program is distributed in the hope that it will be useful,
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
// GNU Affero General Public License for more details.
|
|
//
|
|
// You should have received a copy of the GNU Affero General Public License
|
|
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
package transport
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"net/url"
|
|
"sync"
|
|
|
|
"codeberg.org/gruf/go-byteutil"
|
|
apiutil "github.com/superseriousbusiness/gotosocial/internal/api/util"
|
|
"github.com/superseriousbusiness/gotosocial/internal/config"
|
|
"github.com/superseriousbusiness/gotosocial/internal/gtserror"
|
|
)
|
|
|
|
func (t *transport) BatchDeliver(ctx context.Context, b []byte, recipients []*url.URL) error {
|
|
var (
|
|
// errs accumulates errors received during
|
|
// attempted delivery by deliverer routines.
|
|
errs gtserror.MultiError
|
|
|
|
// wait blocks until all sender
|
|
// routines have returned.
|
|
wait sync.WaitGroup
|
|
|
|
// mutex protects 'recipients' and
|
|
// 'errs' for concurrent access.
|
|
mutex sync.Mutex
|
|
|
|
// Get current instance host info.
|
|
domain = config.GetAccountDomain()
|
|
host = config.GetHost()
|
|
)
|
|
|
|
// Block on expect no. senders.
|
|
wait.Add(t.controller.senders)
|
|
|
|
for i := 0; i < t.controller.senders; i++ {
|
|
go func() {
|
|
// Mark returned.
|
|
defer wait.Done()
|
|
|
|
for {
|
|
// Acquire lock.
|
|
mutex.Lock()
|
|
|
|
if len(recipients) == 0 {
|
|
// Reached end.
|
|
mutex.Unlock()
|
|
return
|
|
}
|
|
|
|
// Pop next recipient.
|
|
i := len(recipients) - 1
|
|
to := recipients[i]
|
|
recipients = recipients[:i]
|
|
|
|
// Done with lock.
|
|
mutex.Unlock()
|
|
|
|
// Skip delivery to recipient if it is "us".
|
|
if to.Host == host || to.Host == domain {
|
|
continue
|
|
}
|
|
|
|
// Attempt to deliver data to recipient.
|
|
if err := t.deliver(ctx, b, to); err != nil {
|
|
mutex.Lock() // safely append err to accumulator.
|
|
errs.Appendf("error delivering to %s: %w", to, err)
|
|
mutex.Unlock()
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
// Wait for finish.
|
|
wait.Wait()
|
|
|
|
// Return combined err.
|
|
return errs.Combine()
|
|
}
|
|
|
|
func (t *transport) Deliver(ctx context.Context, b []byte, to *url.URL) error {
|
|
// if 'to' host is our own, skip as we don't need to deliver to ourselves...
|
|
if to.Host == config.GetHost() || to.Host == config.GetAccountDomain() {
|
|
return nil
|
|
}
|
|
|
|
// Deliver data to recipient.
|
|
return t.deliver(ctx, b, to)
|
|
}
|
|
|
|
func (t *transport) deliver(ctx context.Context, b []byte, to *url.URL) error {
|
|
url := to.String()
|
|
|
|
// Use rewindable bytes reader for body.
|
|
var body byteutil.ReadNopCloser
|
|
body.Reset(b)
|
|
|
|
req, err := http.NewRequestWithContext(ctx, "POST", url, &body)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
req.Header.Add("Content-Type", string(apiutil.AppActivityLDJSON))
|
|
req.Header.Add("Accept-Charset", "utf-8")
|
|
req.Header.Set("Host", to.Host)
|
|
|
|
rsp, err := t.POST(req, b)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rsp.Body.Close()
|
|
|
|
if code := rsp.StatusCode; code != http.StatusOK &&
|
|
code != http.StatusCreated && code != http.StatusAccepted {
|
|
return gtserror.NewFromResponse(rsp)
|
|
}
|
|
|
|
return nil
|
|
}
|