1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
package proton
import (
"context"
"errors"
"fmt"
"strconv"
"github.com/ProtonMail/gluon/async"
"github.com/ProtonMail/gopenpgp/v2/crypto"
"github.com/bradenaw/juniper/iterator"
"github.com/bradenaw/juniper/parallel"
"github.com/bradenaw/juniper/stream"
"github.com/bradenaw/juniper/xslices"
"github.com/go-resty/resty/v2"
)
const (
// maxImportCount is the maximum number of messages that can be imported in a single request.
maxImportCount = 10
// maxImportSize is the maximum total request size permitted for a single import request.
maxImportSize = 30 * 1024 * 1024
)
var ErrImportEncrypt = errors.New("failed to encrypt message")
var ErrImportSizeExceeded = errors.New("message exceeds maximum import size of 30MB")
func (c *Client) ImportMessages(ctx context.Context, addrKR *crypto.KeyRing, workers, buffer int, req ...ImportReq) (stream.Stream[ImportRes], error) {
// Encrypt each message.
for idx := range req {
enc, err := EncryptRFC822(addrKR, req[idx].Message)
if err != nil {
return nil, fmt.Errorf("%w %v: %v", ErrImportEncrypt, idx, err)
}
req[idx].Message = enc
}
// If any of the messages exceed the maximum import size, return an error.
if xslices.Any(req, func(req ImportReq) bool { return len(req.Message) > maxImportSize }) {
return nil, ErrImportSizeExceeded
}
return stream.Flatten(parallel.MapStream(
ctx,
stream.FromIterator(iterator.Slice(chunkSized(req, maxImportCount, maxImportSize, func(req ImportReq) int {
return len(req.Message)
}))),
workers,
buffer,
func(ctx context.Context, req []ImportReq) (stream.Stream[ImportRes], error) {
defer async.HandlePanic(c.m.panicHandler)
res, err := c.importMessages(ctx, req)
if err != nil {
return nil, fmt.Errorf("failed to import messages: %w", err)
}
for _, res := range res {
if res.Code != SuccessCode {
return nil, fmt.Errorf("failed to import message: %w", res.APIError)
}
}
return stream.FromIterator(iterator.Slice(res)), nil
},
)), nil
}
func (c *Client) importMessages(ctx context.Context, req []ImportReq) ([]ImportRes, error) {
names := iterator.Collect(iterator.Map(iterator.Counter(len(req)), func(i int) string {
return strconv.Itoa(i)
}))
var named []namedImportReq
for idx, name := range names {
named = append(named, namedImportReq{
ImportReq: req[idx],
Name: name,
})
}
type namedImportRes struct {
Name string
Response ImportRes
}
var res struct {
Responses []namedImportRes
}
if err := c.do(ctx, func(r *resty.Request) (*resty.Response, error) {
fields, err := buildImportReqFields(named)
if err != nil {
return nil, err
}
return r.SetMultipartFields(fields...).SetResult(&res).Post("/mail/v4/messages/import")
}); err != nil {
return nil, err
}
namedRes := make(map[string]ImportRes, len(res.Responses))
for _, res := range res.Responses {
namedRes[res.Name] = res.Response
}
return xslices.Map(names, func(name string) ImportRes {
return namedRes[name]
}), nil
}
// chunkSized splits a slice into chunks of maximum size and length.
// It is assumed that the size of each element is less than the maximum size.
func chunkSized[T any](vals []T, maxLen, maxSize int, getSize func(T) int) [][]T {
var chunks [][]T
for len(vals) > 0 {
var (
curChunk []T
curSize int
)
for len(vals) > 0 && len(curChunk) < maxLen && curSize < maxSize {
val, size := vals[0], getSize(vals[0])
if curSize+size > maxSize {
break
}
curChunk = append(curChunk, val)
curSize += size
vals = vals[1:]
}
chunks = append(chunks, curChunk)
}
return chunks
}
|
