-
Notifications
You must be signed in to change notification settings - Fork 0
/
mappers.go
56 lines (51 loc) · 2.12 KB
/
mappers.go
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
package contract
import (
"github.com/warp-contracts/syncer/src/utils/arweave"
"github.com/warp-contracts/syncer/src/utils/config"
"github.com/warp-contracts/syncer/src/utils/model"
"github.com/warp-contracts/syncer/src/utils/task"
)
func redisMapper(config *config.Config) (self *task.Mapper[*ContractData, *model.ContractNotification]) {
return task.NewMapper[*ContractData, *model.ContractNotification](config, "map-redis-notification").
WithWorkerPool(1, config.Contract.StoreBatchSize).
WithProcessFunc(func(data *ContractData, out chan *model.ContractNotification) (err error) {
// Neglect messages that are too big
if len(data.Contract.InitState.Bytes) > self.Config.Contract.PublisherMaxMessageSize {
self.Log.WithField("contract_id", data.Contract.ContractId).
WithField("len", len(data.Contract.InitState.Bytes)).
Warn("Init state too big for notifications, skipping")
return err
}
select {
case <-self.Ctx.Done():
case out <- &model.ContractNotification{
ContractTxId: data.Contract.ContractId,
Test: false,
Source: "warp-gw", // FIXME: Should this be changed to another name?
InitialState: data.Contract.InitState,
Tags: []arweave.Tag{}, // Empty array as in the GW
SrcTxId: data.Contract.SrcTxId.String,
}:
}
return nil
})
}
func appSyncMapper(config *config.Config) (self *task.Mapper[*ContractData, *model.AppSyncContractNotification]) {
return task.NewMapper[*ContractData, *model.AppSyncContractNotification](config, "map-appsync-notification").
WithWorkerPool(1, config.Contract.StoreBatchSize).
WithProcessFunc(func(data *ContractData, out chan *model.AppSyncContractNotification) (err error) {
select {
case <-self.Ctx.Done():
case out <- &model.AppSyncContractNotification{
ContractTxId: data.Contract.ContractId,
Source: "arweave",
BlockHeight: data.Contract.BlockHeight,
BlockTimestamp: data.Contract.BlockTimestamp,
Creator: data.Contract.Owner.String,
Type: data.Contract.Type.String,
SyncTimestamp: data.Contract.SyncTimestamp.Int,
}:
}
return nil
})
}