mats/go-cart-actor
1
0
Fork 0
Code Issues Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

refactor/serializing (#1)

Browse Source 
Co-authored-by: matst80 <mats.tornberg@gmail.com>
Reviewed-on: https://git.tornberg.me/mats/go-cart-actor/pulls/1
This commit was merged in pull request #1.
This commit is contained in:
mats
2024-11-08 23:02:09 +01:00
parent 4cc41bcec6
commit 1b75f81119
19 changed files with 576 additions and 409 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
cart-grain.go
View File

@@ -1,95 +1,29 @@
package main package main
import ( import (
"bufio"
"bytes"
"encoding/binary"
"encoding/json" "encoding/json"
"fmt" "fmt"
"io"
"log" "log"
"net/http" "net/http"
"time" "time"
"git.tornberg.me/go-cart-actor/git.tornberg.me/go-cart-actor/messages" messages "git.tornberg.me/go-cart-actor/proto"
"github.com/matst80/slask-finder/pkg/index" "github.com/matst80/slask-finder/pkg/index"
"google.golang.org/protobuf/proto"
) )
type StorableMessage interface { type CartId [16]byte
GetBytes() ([]byte, error)
FromReader(io.Reader, *Message) error func (id CartId) MarshalJSON() ([]byte, error) {
return json.Marshal(id.String())
} }
type Message struct { func (id *CartId) UnmarshalJSON(data []byte) error {
Type uint64 var str string
TimeStamp *int64 err := json.Unmarshal(data, &str)
Content interface{}
}
func (m Message) GetBytes() ([]byte, error) {
var b bytes.Buffer
var err error
w := bufio.NewWriter(&b)
bytes := make([]byte, 8)
binary.LittleEndian.PutUint64(bytes, m.Type)
w.Write(bytes)
binary.LittleEndian.PutUint64(bytes, uint64(*m.TimeStamp))
w.Write(bytes)
var messageBytes []byte
if m.Type == 1 {
messageBytes, err = proto.Marshal(m.Content.(*messages.AddRequest))
} else if m.Type == 2 {
messageBytes, err = proto.Marshal(m.Content.(*messages.AddItem))
} else {
return nil, fmt.Errorf("unknown message type")
}
if err != nil {
return nil, err
}
binary.LittleEndian.PutUint64(bytes, uint64(len(messageBytes)))
w.Write(bytes)
w.Write(messageBytes)
w.Flush()
return b.Bytes(), nil
}
func (i Message) FromReader(reader io.Reader, m *Message) error {
bytes := make([]byte, 8)
if _, err := reader.Read(bytes); err != nil {
return err
}
m.Type = binary.LittleEndian.Uint64(bytes)
if _, err := reader.Read(bytes); err != nil {
return err
}
timestamp := int64(binary.LittleEndian.Uint64(bytes))
m.TimeStamp = &timestamp
if _, err := reader.Read(bytes); err != nil {
return err
}
messageBytes := make([]byte, binary.LittleEndian.Uint64(bytes))
if _, err := reader.Read(messageBytes); err != nil {
return err
}
var err error
if m.Type == 1 {
msg := &messages.AddRequest{}
err = proto.Unmarshal(messageBytes, msg)
m.Content = msg
} else if m.Type == 2 {
msg := &messages.AddItem{}
err = proto.Unmarshal(messageBytes, msg)
m.Content = msg
} else {
return fmt.Errorf("unknown message type")
}
if err != nil { if err != nil {
return err return err
} }
copy(id[:], []byte(str))
return nil return nil
} }
@@ -102,19 +36,17 @@ type CartItem struct {
type CartGrain struct { type CartGrain struct {
storageMessages []Message storageMessages []Message
Id string `json:"id"` Id CartId `json:"id"`
Items []CartItem `json:"items"` Items []CartItem `json:"items"`
TotalPrice int64 `json:"totalPrice"` TotalPrice int64 `json:"totalPrice"`
} }
type Grain interface { type Grain interface {
GetId() string GetId() CartId
GetLastChange() int64 HandleMessage(message *Message, isReplay bool) ([]byte, error)
HandleMessage(message *Message, isReplay bool, reply *CartGrain) error
GetStorageMessage(since int64) []StorableMessage
} }
func (c *CartGrain) GetId() string { func (c *CartGrain) GetId() CartId {
return c.Id return c.Id
} }
@@ -141,14 +73,14 @@ func getItemData(sku string) (*messages.AddItem, error) {
priceField, ok := item.GetFields()[4] priceField, ok := item.GetFields()[4]
if ok { if ok {
pricef, ok := priceField.(float64) priceFloat, ok := priceField.(float64)
if !ok { if !ok {
price, ok = priceField.(int) price, ok = priceField.(int)
if !ok { if !ok {
return nil, fmt.Errorf("invalid price type") return nil, fmt.Errorf("invalid price type")
} }
} else { } else {
price = int(pricef) price = int(priceFloat)
} }
} }
} }
@@ -161,15 +93,15 @@ func getItemData(sku string) (*messages.AddItem, error) {
}, nil }, nil
} }
func (c *CartGrain) AddItem(sku string, reply *CartGrain) error { func (c *CartGrain) AddItem(sku string) ([]byte, error) {
cartItem, err := getItemData(sku) cartItem, err := getItemData(sku)
if err != nil { if err != nil {
return err return nil, err
} }
return c.HandleMessage(&Message{ return c.HandleMessage(&Message{
Type: 2, Type: 2,
Content: cartItem, Content: cartItem,
}, false, reply) }, false)
} }
func (c *CartGrain) GetStorageMessage(since int64) []StorableMessage { func (c *CartGrain) GetStorageMessage(since int64) []StorableMessage {
@@ -183,7 +115,7 @@ func (c *CartGrain) GetStorageMessage(since int64) []StorableMessage {
return ret return ret
} }
func (c *CartGrain) HandleMessage(message *Message, isReplay bool, reply *CartGrain) error { func (c *CartGrain) HandleMessage(message *Message, isReplay bool) ([]byte, error) {
log.Printf("Handling message %d", message.Type) log.Printf("Handling message %d", message.Type)
if message.TimeStamp == nil { if message.TimeStamp == nil {
now := time.Now().Unix() now := time.Now().Unix()
@@ -191,14 +123,14 @@ func (c *CartGrain) HandleMessage(message *Message, isReplay bool, reply *CartGr
} }
var err error var err error
switch message.Type { switch message.Type {
case 1: case AddRequestType:
msg, ok := message.Content.(*messages.AddRequest) msg, ok := message.Content.(*messages.AddRequest)
if !ok { if !ok {
err = fmt.Errorf("invalid content type") err = fmt.Errorf("invalid content type")
} else { } else {
return c.AddItem(msg.Sku, reply) return c.AddItem(msg.Sku)
} }
case 2: case AddItemType:
msg, ok := message.Content.(*messages.AddItem) msg, ok := message.Content.(*messages.AddItem)
if !ok { if !ok {
err = fmt.Errorf("invalid content type") err = fmt.Errorf("invalid content type")
@@ -214,9 +146,11 @@ func (c *CartGrain) HandleMessage(message *Message, isReplay bool, reply *CartGr
default: default:
err = fmt.Errorf("unknown message type") err = fmt.Errorf("unknown message type")
} }
if err != nil {
return nil, err
}
if !isReplay { if !isReplay {
c.storageMessages = append(c.storageMessages, *message) c.storageMessages = append(c.storageMessages, *message)
} }
*reply = *c return json.Marshal(c)
return err
} }

BIN
data/1.prot
View File

Binary file not shown.

BIN
data/4.prot Normal file
View File

Binary file not shown.

BIN
data/state.gob Normal file
View File

Binary file not shown.

BIN
data/state.gob.bak Normal file
View File

Binary file not shown.

1
data/state.json
View File

@@ -1 +0,0 @@
{"1":1731051371}

1
data/state.json.bak
View File

@@ -1 +0,0 @@
{"1":1731050604}

33
disk-storage.go
View File

@@ -1,7 +1,7 @@
package main package main
import ( import (
"encoding/json" "encoding/gob"
"errors" "errors"
"fmt" "fmt"
"log" "log"
@@ -12,26 +12,26 @@ import (
type DiskStorage struct { type DiskStorage struct {
stateFile string stateFile string
lastSave int64 lastSave int64
LastSaves map[string]int64 LastSaves map[CartId]int64
} }
func NewDiskStorage(stateFile string) (*DiskStorage, error) { func NewDiskStorage(stateFile string) (*DiskStorage, error) {
ret := &DiskStorage{ ret := &DiskStorage{
stateFile: stateFile, stateFile: stateFile,
LastSaves: make(map[string]int64), LastSaves: make(map[CartId]int64),
} }
err := ret.loadState() err := ret.loadState()
return ret, err return ret, err
} }
func saveMessages(messages []StorableMessage, id string) error { func saveMessages(messages []StorableMessage, id CartId) error {
log.Printf("%d messages to save for %s", len(messages), id) log.Printf("%d messages to save for %s", len(messages), id)
if len(messages) == 0 { if len(messages) == 0 {
return nil return nil
} }
var file *os.File var file *os.File
var err error var err error
path := getCartPath(id) path := getCartPath(id.String())
file, err = os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) file, err = os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil { if err != nil {
return err return err
@@ -39,11 +39,10 @@ func saveMessages(messages []StorableMessage, id string) error {
defer file.Close() defer file.Close()
for _, m := range messages { for _, m := range messages {
b, err := m.GetBytes() err := m.Write(file)
if err != nil { if err != nil {
return err return err
} }
file.Write(b)
} }
return err return err
} }
@@ -52,9 +51,9 @@ func getCartPath(id string) string {
return fmt.Sprintf("data/%s.prot", id) return fmt.Sprintf("data/%s.prot", id)
} }
func loadMessages(grain Grain, id string) error { func loadMessages(grain Grain, id CartId) error {
var err error var err error
path := getCartPath(id) path := getCartPath(id.String())
if _, err = os.Stat(path); errors.Is(err, os.ErrNotExist) { if _, err = os.Stat(path); errors.Is(err, os.ErrNotExist) {
return err return err
} }
@@ -64,13 +63,11 @@ func loadMessages(grain Grain, id string) error {
} }
defer file.Close() defer file.Close()
var reply CartGrain
for err == nil { for err == nil {
msg := &Message{} var msg Message
err = msg.FromReader(file, msg) err = MessageFromReader(file, &msg)
if err == nil { if err == nil {
grain.HandleMessage(msg, true, &reply) grain.HandleMessage(&msg, true)
} }
} }
@@ -87,7 +84,7 @@ func (s *DiskStorage) saveState() error {
return err return err
} }
defer file.Close() defer file.Close()
err = json.NewEncoder(file).Encode(s.LastSaves) err = gob.NewEncoder(file).Encode(s.LastSaves)
if err != nil { if err != nil {
return err return err
} }
@@ -97,15 +94,15 @@ func (s *DiskStorage) saveState() error {
} }
func (s *DiskStorage) loadState() error { func (s *DiskStorage) loadState() error {
file, err := os.Open("data/state.json") file, err := os.Open(s.stateFile)
if err != nil { if err != nil {
return err return err
} }
defer file.Close() defer file.Close()
return json.NewDecoder(file).Decode(&s.LastSaves) return gob.NewDecoder(file).Decode(&s.LastSaves)
} }
func (s *DiskStorage) Store(id string, grain Grain) error { func (s *DiskStorage) Store(id CartId, grain *CartGrain) error {
lastSavedMessage, ok := s.LastSaves[id] lastSavedMessage, ok := s.LastSaves[id]
if ok && lastSavedMessage > grain.GetLastChange() { if ok && lastSavedMessage > grain.GetLastChange() {
return nil return nil

212
git.tornberg.me/go-cart-actor/add.pb.go
View File

@@ -1,212 +0,0 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.35.1
// protoc v5.28.2
// source: add.proto
package go_cart_actor
import (
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
type AddRequest struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Sku string `protobuf:"bytes,2,opt,name=Sku,proto3" json:"Sku,omitempty"`
}
func (x *AddRequest) Reset() {
*x = AddRequest{}
mi := &file_add_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *AddRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*AddRequest) ProtoMessage() {}
func (x *AddRequest) ProtoReflect() protoreflect.Message {
mi := &file_add_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use AddRequest.ProtoReflect.Descriptor instead.
func (*AddRequest) Descriptor() ([]byte, []int) {
return file_add_proto_rawDescGZIP(), []int{0}
}
func (x *AddRequest) GetSku() string {
if x != nil {
return x.Sku
}
return ""
}
type AddItem struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Quantity int32 `protobuf:"varint,2,opt,name=Quantity,proto3" json:"Quantity,omitempty"`
Price int64 `protobuf:"varint,3,opt,name=Price,proto3" json:"Price,omitempty"`
Sku string `protobuf:"bytes,4,opt,name=Sku,proto3" json:"Sku,omitempty"`
Name string `protobuf:"bytes,5,opt,name=Name,proto3" json:"Name,omitempty"`
Image string `protobuf:"bytes,6,opt,name=Image,proto3" json:"Image,omitempty"`
}
func (x *AddItem) Reset() {
*x = AddItem{}
mi := &file_add_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *AddItem) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*AddItem) ProtoMessage() {}
func (x *AddItem) ProtoReflect() protoreflect.Message {
mi := &file_add_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use AddItem.ProtoReflect.Descriptor instead.
func (*AddItem) Descriptor() ([]byte, []int) {
return file_add_proto_rawDescGZIP(), []int{1}
}
func (x *AddItem) GetQuantity() int32 {
if x != nil {
return x.Quantity
}
return 0
}
func (x *AddItem) GetPrice() int64 {
if x != nil {
return x.Price
}
return 0
}
func (x *AddItem) GetSku() string {
if x != nil {
return x.Sku
}
return ""
}
func (x *AddItem) GetName() string {
if x != nil {
return x.Name
}
return ""
}
func (x *AddItem) GetImage() string {
if x != nil {
return x.Image
}
return ""
}
var File_add_proto protoreflect.FileDescriptor
var file_add_proto_rawDesc = []byte{
0x0a, 0x09, 0x61, 0x64, 0x64, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x08, 0x6d, 0x65, 0x73,
0x73, 0x61, 0x67, 0x65, 0x73, 0x22, 0x1e, 0x0a, 0x0a, 0x41, 0x64, 0x64, 0x52, 0x65, 0x71, 0x75,
0x65, 0x73, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x53, 0x6b, 0x75, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
0x52, 0x03, 0x53, 0x6b, 0x75, 0x22, 0x77, 0x0a, 0x07, 0x41, 0x64, 0x64, 0x49, 0x74, 0x65, 0x6d,
0x12, 0x1a, 0x0a, 0x08, 0x51, 0x75, 0x61, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x18, 0x02, 0x20, 0x01,
0x28, 0x05, 0x52, 0x08, 0x51, 0x75, 0x61, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x12, 0x14, 0x0a, 0x05,
0x50, 0x72, 0x69, 0x63, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x03, 0x52, 0x05, 0x50, 0x72, 0x69,
0x63, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x53, 0x6b, 0x75, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52,
0x03, 0x53, 0x6b, 0x75, 0x12, 0x12, 0x0a, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01,
0x28, 0x09, 0x52, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x49, 0x6d, 0x61, 0x67,
0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x49, 0x6d, 0x61, 0x67, 0x65, 0x42, 0x1f,
0x5a, 0x1d, 0x67, 0x69, 0x74, 0x2e, 0x74, 0x6f, 0x72, 0x6e, 0x62, 0x65, 0x72, 0x67, 0x2e, 0x6d,
0x65, 0x2f, 0x67, 0x6f, 0x2d, 0x63, 0x61, 0x72, 0x74, 0x2d, 0x61, 0x63, 0x74, 0x6f, 0x72, 0x62,
0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
file_add_proto_rawDescOnce sync.Once
file_add_proto_rawDescData = file_add_proto_rawDesc
)
func file_add_proto_rawDescGZIP() []byte {
file_add_proto_rawDescOnce.Do(func() {
file_add_proto_rawDescData = protoimpl.X.CompressGZIP(file_add_proto_rawDescData)
})
return file_add_proto_rawDescData
}
var file_add_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
var file_add_proto_goTypes = []any{
(*AddRequest)(nil), // 0: messages.AddRequest
(*AddItem)(nil), // 1: messages.AddItem
}
var file_add_proto_depIdxs = []int32{
0, // [0:0] is the sub-list for method output_type
0, // [0:0] is the sub-list for method input_type
0, // [0:0] is the sub-list for extension type_name
0, // [0:0] is the sub-list for extension extendee
0, // [0:0] is the sub-list for field type_name
}
func init() { file_add_proto_init() }
func file_add_proto_init() {
if File_add_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_add_proto_rawDesc,
NumEnums: 0,
NumMessages: 2,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_add_proto_goTypes,
DependencyIndexes: file_add_proto_depIdxs,
MessageInfos: file_add_proto_msgTypes,
}.Build()
File_add_proto = out.File
file_add_proto_rawDesc = nil
file_add_proto_goTypes = nil
file_add_proto_depIdxs = nil
}

40
grain-pool.go
View File

@@ -7,25 +7,27 @@ import (
) )
type GrainPool interface { type GrainPool interface {
GetOrSpawn(id string) (Grain, error) Process(id CartId, messages ...Message) (interface{}, error)
Get(id string) (Grain, error) Get(id CartId) (Grain, error)
} }
type Ttl struct { type Ttl struct {
Expires time.Time Expires time.Time
Item Grain Item *CartGrain
} }
type GrainLocalPool struct { type GrainLocalPool struct {
grains map[string]Grain grains map[CartId]*CartGrain
expiry []Ttl expiry []Ttl
spawn func(id CartId) (*CartGrain, error)
Ttl time.Duration Ttl time.Duration
PoolSize int PoolSize int
} }
func NewGrainLocalPool(size int, ttl time.Duration) *GrainLocalPool { func NewGrainLocalPool(size int, ttl time.Duration, spawn func(id CartId) (*CartGrain, error)) *GrainLocalPool {
ret := &GrainLocalPool{ ret := &GrainLocalPool{
grains: make(map[string]Grain), spawn: spawn,
grains: make(map[CartId]*CartGrain),
expiry: make([]Ttl, 0), expiry: make([]Ttl, 0),
Ttl: ttl, Ttl: ttl,
PoolSize: size, PoolSize: size,
@@ -59,11 +61,12 @@ func (p *GrainLocalPool) Purge() {
} }
} }
func (p *GrainLocalPool) GetGrains() map[string]Grain { func (p *GrainLocalPool) GetGrains() map[CartId]*CartGrain {
return p.grains return p.grains
} }
func (p *GrainLocalPool) GetOrSpawn(id string, generator func(id string) Grain) (Grain, error) { func (p *GrainLocalPool) GetGrain(id CartId) (*CartGrain, error) {
var err error
grain, ok := p.grains[id] grain, ok := p.grains[id]
if !ok { if !ok {
if len(p.grains) >= p.PoolSize { if len(p.grains) >= p.PoolSize {
@@ -74,16 +77,23 @@ func (p *GrainLocalPool) GetOrSpawn(id string, generator func(id string) Grain)
return nil, fmt.Errorf("pool is full") return nil, fmt.Errorf("pool is full")
} }
} }
grain = generator(id) grain, err = p.spawn(id)
p.grains[id] = grain p.grains[id] = grain
} }
return grain, nil return grain, err
} }
func (p *GrainLocalPool) Get(id string) (Grain, error) { func (p *GrainLocalPool) Process(id CartId, messages ...Message) (interface{}, error) {
grain, ok := p.grains[id] grain, err := p.GetGrain(id)
if !ok { if err == nil && grain != nil {
return nil, fmt.Errorf("grain not found") for _, message := range messages {
_, err = grain.HandleMessage(&message, false)
}
} }
return grain, nil return grain, err
}
func (p *GrainLocalPool) Get(id CartId) (Grain, error) {
return p.GetGrain(id)
} }

76
main.go
View File

@@ -1,18 +1,16 @@
package main package main
import ( import (
"encoding/gob"
"encoding/json" "encoding/json"
"log" "log"
"net/http" "net/http"
"os" "os"
"time" "time"
"git.tornberg.me/go-cart-actor/git.tornberg.me/go-cart-actor/messages" messages "git.tornberg.me/go-cart-actor/proto"
) )
func spawn(id string) Grain { func spawn(id CartId) (*CartGrain, error) {
ret := &CartGrain{ ret := &CartGrain{
Id: id, Id: id,
Items: []CartItem{}, Items: []CartItem{},
@@ -20,16 +18,11 @@ func spawn(id string) Grain {
TotalPrice: 0, TotalPrice: 0,
} }
err := loadMessages(ret, id) err := loadMessages(ret, id)
if err != nil { return ret, err
log.Printf("Error loading messages for grain %s: %v\n", id, err)
}
return ret
} }
func init() { func init() {
os.Mkdir("data", 0755) os.Mkdir("data", 0755)
gob.Register(CartItem{})
gob.Register(Message{})
} }
type App struct { type App struct {
@@ -39,7 +32,7 @@ type App struct {
func (a *App) HandleGet(w http.ResponseWriter, r *http.Request) { func (a *App) HandleGet(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id") id := r.PathValue("id")
grain, err := a.pool.GetOrSpawn(id, spawn) grain, err := a.pool.Get(ToCartId(id))
if err != nil { if err != nil {
w.WriteHeader(http.StatusNotFound) w.WriteHeader(http.StatusNotFound)
return return
@@ -52,25 +45,19 @@ func (a *App) HandleGet(w http.ResponseWriter, r *http.Request) {
func (a *App) HandleAddSku(w http.ResponseWriter, r *http.Request) { func (a *App) HandleAddSku(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id") id := r.PathValue("id")
sku := r.PathValue("sku") sku := r.PathValue("sku")
grain, err := a.pool.GetOrSpawn(id, spawn) grain, err := a.pool.Process(ToCartId(id), Message{
if err != nil { Type: AddRequestType,
w.WriteHeader(http.StatusNotFound)
return
}
message := &Message{
Type: 1,
Content: &messages.AddRequest{Sku: sku}, Content: &messages.AddRequest{Sku: sku},
} })
var reply CartGrain
err = grain.HandleMessage(message, false, &reply)
if err != nil { if err != nil {
w.WriteHeader(http.StatusInternalServerError) w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error())) w.Write([]byte(err.Error()))
return return
} }
w.Header().Set("Content-Type", "application/json") w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(reply) json.NewEncoder(w).Encode(grain)
} }
func (a *App) Save() error { func (a *App) Save() error {
@@ -95,18 +82,55 @@ func (a *App) HandleSave(w http.ResponseWriter, r *http.Request) {
func main() { func main() {
// Create a new instance of the server // Create a new instance of the server
storage, err := NewDiskStorage("data/state.json") storage, err := NewDiskStorage("data/state.gob")
if err != nil { if err != nil {
log.Printf("Error loading state: %v\n", err) log.Printf("Error loading state: %v\n", err)
} }
app := &App{ app := &App{
pool: NewGrainLocalPool(1000, 5*time.Minute), pool: NewGrainLocalPool(1000, 5*time.Minute, spawn),
storage: storage, storage: storage,
} }
rpcHandler, err := NewGrainHandler(app.pool, "localhost:1337")
if err != nil {
log.Fatalf("Error creating handler: %v\n", err)
}
go rpcHandler.Serve()
remotePool := NewRemoteGrainPool("localhost:1337")
mux := http.NewServeMux() mux := http.NewServeMux()
mux.HandleFunc("GET /{id}", app.HandleGet) mux.HandleFunc("GET /api/{id}", app.HandleGet)
mux.HandleFunc("GET /{id}/add/{sku}", app.HandleAddSku) mux.HandleFunc("GET /api/{id}/add/{sku}", app.HandleAddSku)
mux.HandleFunc("GET /remote/{id}/add", func(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
ts := time.Now().Unix()
data, err := remotePool.Process(ToCartId(id), Message{
Type: AddRequestType,
TimeStamp: &ts,
Content: &messages.AddRequest{Sku: "49565"},
})
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error()))
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
w.Write(data)
})
mux.HandleFunc("GET /remote/{id}", func(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
data, err := remotePool.Get(ToCartId(id))
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error()))
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
w.Write(data)
})
mux.HandleFunc("GET /save", app.HandleSave) mux.HandleFunc("GET /save", app.HandleSave)
http.ListenAndServe(":8080", mux) http.ListenAndServe(":8080", mux)

6
message-types.go Normal file
View File

@@ -0,0 +1,6 @@
package main
const (
AddRequestType = 1
AddItemType = 2
)

111
message.go Normal file
View File

@@ -0,0 +1,111 @@
package main
import (
"bytes"
"encoding/binary"
"fmt"
"io"
"time"
messages "git.tornberg.me/go-cart-actor/proto"
"google.golang.org/protobuf/proto"
)
type StorableMessage interface {
Write(w io.Writer) error
}
type Message struct {
Type uint16
TimeStamp *int64
Content interface{}
}
type MessageWriter struct {
io.Writer
}
type StorableMessageHeader struct {
Version uint16
Type uint16
TimeStamp int64
DataLength uint64
}
func GetData(fn func(w io.Writer) error) ([]byte, error) {
var buf bytes.Buffer
err := fn(&buf)
if err != nil {
return nil, err
}
b := buf.Bytes()
return b, nil
}
func (m Message) Write(w io.Writer) error {
data, err := GetData(func(wr io.Writer) error {
if m.Type == AddRequestType {
messageBytes, err := proto.Marshal(m.Content.(*messages.AddRequest))
if err != nil {
return err
}
wr.Write(messageBytes)
} else if m.Type == AddItemType {
messageBytes, err := proto.Marshal(m.Content.(*messages.AddItem))
if err != nil {
return err
}
wr.Write(messageBytes)
}
return nil
})
if err != nil {
return err
}
ts := time.Now().Unix()
if m.TimeStamp != nil {
ts = *m.TimeStamp
}
err = binary.Write(w, binary.LittleEndian, StorableMessageHeader{
Version: 1,
Type: m.Type,
TimeStamp: ts,
DataLength: uint64(len(data)),
})
w.Write(data)
return err
}
func MessageFromReader(reader io.Reader, m *Message) error {
header := StorableMessageHeader{}
err := binary.Read(reader, binary.LittleEndian, &header)
if err != nil {
return err
}
messageBytes := make([]byte, header.DataLength)
_, err = reader.Read(messageBytes)
if err != nil {
return err
}
switch header.Type {
case AddRequestType:
msg := &messages.AddRequest{}
err = proto.Unmarshal(messageBytes, msg)
m.Content = msg
case AddItemType:
msg := &messages.AddItem{}
err = proto.Unmarshal(messageBytes, msg)
m.Content = msg
default:
return fmt.Errorf("unknown message type")
}
if err != nil {
return err
}
m.Type = header.Type
m.TimeStamp = &header.TimeStamp
return nil
}

76
packet.go Normal file
View File

@@ -0,0 +1,76 @@
package main
import (
"encoding/binary"
"io"
)
const (
RemoteGetState = uint16(0x01)
RemoteHandleMessage = uint16(0x02)
ResponseBody = uint16(0x03)
)
type CartPacket struct {
Version uint16
MessageType uint16
Id CartId
DataLength uint16
}
type ResponsePacket struct {
Version uint16
MessageType uint16
DataLength uint16
}
func SendCartPacket(conn io.Writer, id CartId, messageType uint16, datafn func(w io.Writer) error) error {
data, err := GetData(datafn)
if err != nil {
return err
}
binary.Write(conn, binary.LittleEndian, CartPacket{
Version: 2,
MessageType: messageType,
Id: id,
DataLength: uint16(len(data)),
})
_, err = conn.Write(data)
return err
}
func SendPacket(conn io.Writer, messageType uint16, datafn func(w io.Writer) error) error {
data, err := GetData(datafn)
if err != nil {
return err
}
binary.Write(conn, binary.LittleEndian, ResponsePacket{
Version: 1,
MessageType: messageType,
DataLength: uint16(len(data)),
})
_, err = conn.Write(data)
return err
}
// func ReceiveCartPacket(conn io.Reader) (CartPacket, []byte, error) {
// var packet CartPacket
// err := binary.Read(conn, binary.LittleEndian, &packet)
// if err != nil {
// return packet, nil, err
// }
// data := make([]byte, packet.DataLength)
// _, err = conn.Read(data)
// return packet, data, err
// }
func ReceivePacket(conn io.Reader) (uint16, []byte, error) {
var packet ResponsePacket
err := binary.Read(conn, binary.LittleEndian, &packet)
if err != nil {
return packet.MessageType, nil, err
}
data := make([]byte, packet.DataLength)
_, err = conn.Read(data)
return packet.MessageType, data, err
}

85
git.tornberg.me/go-cart-actor/messages/add.pb.go → proto/messages.pb.go
View File

@@ -2,7 +2,7 @@
// versions: // versions:
// protoc-gen-go v1.35.1 // protoc-gen-go v1.35.1
// protoc v5.28.2 // protoc v5.28.2
// source: add.proto // source: proto/messages.proto
package messages package messages
@@ -30,7 +30,7 @@ type AddRequest struct {
func (x *AddRequest) Reset() { func (x *AddRequest) Reset() {
*x = AddRequest{} *x = AddRequest{}
mi := &file_add_proto_msgTypes[0] mi := &file_proto_messages_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi) ms.StoreMessageInfo(mi)
} }
@@ -42,7 +42,7 @@ func (x *AddRequest) String() string {
func (*AddRequest) ProtoMessage() {} func (*AddRequest) ProtoMessage() {}
func (x *AddRequest) ProtoReflect() protoreflect.Message { func (x *AddRequest) ProtoReflect() protoreflect.Message {
mi := &file_add_proto_msgTypes[0] mi := &file_proto_messages_proto_msgTypes[0]
if x != nil { if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil { if ms.LoadMessageInfo() == nil {
@@ -55,7 +55,7 @@ func (x *AddRequest) ProtoReflect() protoreflect.Message {
// Deprecated: Use AddRequest.ProtoReflect.Descriptor instead. // Deprecated: Use AddRequest.ProtoReflect.Descriptor instead.
func (*AddRequest) Descriptor() ([]byte, []int) { func (*AddRequest) Descriptor() ([]byte, []int) {
return file_add_proto_rawDescGZIP(), []int{0} return file_proto_messages_proto_rawDescGZIP(), []int{0}
} }
func (x *AddRequest) GetSku() string { func (x *AddRequest) GetSku() string {
@@ -79,7 +79,7 @@ type AddItem struct {
func (x *AddItem) Reset() { func (x *AddItem) Reset() {
*x = AddItem{} *x = AddItem{}
mi := &file_add_proto_msgTypes[1] mi := &file_proto_messages_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi) ms.StoreMessageInfo(mi)
} }
@@ -91,7 +91,7 @@ func (x *AddItem) String() string {
func (*AddItem) ProtoMessage() {} func (*AddItem) ProtoMessage() {}
func (x *AddItem) ProtoReflect() protoreflect.Message { func (x *AddItem) ProtoReflect() protoreflect.Message {
mi := &file_add_proto_msgTypes[1] mi := &file_proto_messages_proto_msgTypes[1]
if x != nil { if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil { if ms.LoadMessageInfo() == nil {
@@ -104,7 +104,7 @@ func (x *AddItem) ProtoReflect() protoreflect.Message {
// Deprecated: Use AddItem.ProtoReflect.Descriptor instead. // Deprecated: Use AddItem.ProtoReflect.Descriptor instead.
func (*AddItem) Descriptor() ([]byte, []int) { func (*AddItem) Descriptor() ([]byte, []int) {
return file_add_proto_rawDescGZIP(), []int{1} return file_proto_messages_proto_rawDescGZIP(), []int{1}
} }
func (x *AddItem) GetQuantity() int32 { func (x *AddItem) GetQuantity() int32 {
@@ -142,43 +142,42 @@ func (x *AddItem) GetImage() string {
return "" return ""
} }
var File_add_proto protoreflect.FileDescriptor var File_proto_messages_proto protoreflect.FileDescriptor
var file_add_proto_rawDesc = []byte{ var file_proto_messages_proto_rawDesc = []byte{
0x0a, 0x09, 0x61, 0x64, 0x64, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x08, 0x6d, 0x65, 0x73, 0x0a, 0x14, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73,
0x73, 0x61, 0x67, 0x65, 0x73, 0x22, 0x1e, 0x0a, 0x0a, 0x41, 0x64, 0x64, 0x52, 0x65, 0x71, 0x75, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73,
0x65, 0x73, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x53, 0x6b, 0x75, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x22, 0x1e, 0x0a, 0x0a, 0x41, 0x64, 0x64, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x10,
0x52, 0x03, 0x53, 0x6b, 0x75, 0x22, 0x77, 0x0a, 0x07, 0x41, 0x64, 0x64, 0x49, 0x74, 0x65, 0x6d, 0x0a, 0x03, 0x53, 0x6b, 0x75, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x53, 0x6b, 0x75,
0x12, 0x1a, 0x0a, 0x08, 0x51, 0x75, 0x61, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x18, 0x02, 0x20, 0x01, 0x22, 0x77, 0x0a, 0x07, 0x41, 0x64, 0x64, 0x49, 0x74, 0x65, 0x6d, 0x12, 0x1a, 0x0a, 0x08, 0x51,
0x28, 0x05, 0x52, 0x08, 0x51, 0x75, 0x61, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x75, 0x61, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x08, 0x51,
0x50, 0x72, 0x69, 0x63, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x03, 0x52, 0x05, 0x50, 0x72, 0x69, 0x75, 0x61, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x50, 0x72, 0x69, 0x63, 0x65,
0x63, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x53, 0x6b, 0x75, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x18, 0x03, 0x20, 0x01, 0x28, 0x03, 0x52, 0x05, 0x50, 0x72, 0x69, 0x63, 0x65, 0x12, 0x10, 0x0a,
0x03, 0x53, 0x6b, 0x75, 0x12, 0x12, 0x0a, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x03, 0x53, 0x6b, 0x75, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x53, 0x6b, 0x75, 0x12,
0x28, 0x09, 0x52, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x49, 0x6d, 0x61, 0x67, 0x12, 0x0a, 0x04, 0x4e, 0x61, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x4e,
0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x49, 0x6d, 0x61, 0x67, 0x65, 0x42, 0x28, 0x61, 0x6d, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x49, 0x6d, 0x61, 0x67, 0x65, 0x18, 0x06, 0x20, 0x01,
0x5a, 0x26, 0x67, 0x69, 0x74, 0x2e, 0x74, 0x6f, 0x72, 0x6e, 0x62, 0x65, 0x72, 0x67, 0x2e, 0x6d, 0x28, 0x09, 0x52, 0x05, 0x49, 0x6d, 0x61, 0x67, 0x65, 0x42, 0x0c, 0x5a, 0x0a, 0x2e, 0x3b, 0x6d,
0x65, 0x2f, 0x67, 0x6f, 0x2d, 0x63, 0x61, 0x72, 0x74, 0x2d, 0x61, 0x63, 0x74, 0x6f, 0x72, 0x2f, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
} }
var ( var (
file_add_proto_rawDescOnce sync.Once file_proto_messages_proto_rawDescOnce sync.Once
file_add_proto_rawDescData = file_add_proto_rawDesc file_proto_messages_proto_rawDescData = file_proto_messages_proto_rawDesc
) )
func file_add_proto_rawDescGZIP() []byte { func file_proto_messages_proto_rawDescGZIP() []byte {
file_add_proto_rawDescOnce.Do(func() { file_proto_messages_proto_rawDescOnce.Do(func() {
file_add_proto_rawDescData = protoimpl.X.CompressGZIP(file_add_proto_rawDescData) file_proto_messages_proto_rawDescData = protoimpl.X.CompressGZIP(file_proto_messages_proto_rawDescData)
}) })
return file_add_proto_rawDescData return file_proto_messages_proto_rawDescData
} }
var file_add_proto_msgTypes = make([]protoimpl.MessageInfo, 2) var file_proto_messages_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
var file_add_proto_goTypes = []any{ var file_proto_messages_proto_goTypes = []any{
(*AddRequest)(nil), // 0: messages.AddRequest (*AddRequest)(nil), // 0: messages.AddRequest
(*AddItem)(nil), // 1: messages.AddItem (*AddItem)(nil), // 1: messages.AddItem
} }
var file_add_proto_depIdxs = []int32{ var file_proto_messages_proto_depIdxs = []int32{
0, // [0:0] is the sub-list for method output_type 0, // [0:0] is the sub-list for method output_type
0, // [0:0] is the sub-list for method input_type 0, // [0:0] is the sub-list for method input_type
0, // [0:0] is the sub-list for extension type_name 0, // [0:0] is the sub-list for extension type_name
@@ -186,27 +185,27 @@ var file_add_proto_depIdxs = []int32{
0, // [0:0] is the sub-list for field type_name 0, // [0:0] is the sub-list for field type_name
} }
func init() { file_add_proto_init() } func init() { file_proto_messages_proto_init() }
func file_add_proto_init() { func file_proto_messages_proto_init() {
if File_add_proto != nil { if File_proto_messages_proto != nil {
return return
} }
type x struct{} type x struct{}
out := protoimpl.TypeBuilder{ out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{ File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(), GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_add_proto_rawDesc, RawDescriptor: file_proto_messages_proto_rawDesc,
NumEnums: 0, NumEnums: 0,
NumMessages: 2, NumMessages: 2,
NumExtensions: 0, NumExtensions: 0,
NumServices: 0, NumServices: 0,
}, },
GoTypes: file_add_proto_goTypes, GoTypes: file_proto_messages_proto_goTypes,
DependencyIndexes: file_add_proto_depIdxs, DependencyIndexes: file_proto_messages_proto_depIdxs,
MessageInfos: file_add_proto_msgTypes, MessageInfos: file_proto_messages_proto_msgTypes,
}.Build() }.Build()
File_add_proto = out.File File_proto_messages_proto = out.File
file_add_proto_rawDesc = nil file_proto_messages_proto_rawDesc = nil
file_add_proto_goTypes = nil file_proto_messages_proto_goTypes = nil
file_add_proto_depIdxs = nil file_proto_messages_proto_depIdxs = nil
} }

2
messages/add.proto → proto/messages.proto
View File

@@ -1,6 +1,6 @@
syntax = "proto3"; syntax = "proto3";
package messages; package messages;
option go_package = "git.tornberg.me/go-cart-actor/messages"; option go_package = ".;messages";
message AddRequest { message AddRequest {
string Sku = 2; string Sku = 2;

0
proto/service.proto Normal file
View File

110
rpc-pool.go Normal file
View File

@@ -0,0 +1,110 @@
package main
import (
"io"
"net"
"strings"
)
type RemoteGrainPool struct {
Hosts []string
grains map[CartId]RemoteGrain
}
func (id CartId) String() string {
return strings.Trim(string(id[:]), "\x00")
}
func ToCartId(id string) CartId {
var result [16]byte
copy(result[:], []byte(id))
return result
}
type RemoteGrain struct {
client net.Conn
Id CartId
Address string
}
func NewRemoteGrain(id CartId, address string) *RemoteGrain {
return &RemoteGrain{
Id: id,
Address: address,
}
}
func (g *RemoteGrain) Connect() error {
if g.client == nil {
client, err := net.Dial("tcp", g.Address)
if err != nil {
return err
}
g.client = client
}
return nil
}
func (g *RemoteGrain) HandleMessage(message *Message, isReplay bool) ([]byte, error) {
err := SendCartPacket(g.client, g.Id, RemoteHandleMessage, message.Write)
if err != nil {
return nil, err
}
_, data, err := ReceivePacket(g.client)
return data, err
}
func (g *RemoteGrain) GetId() CartId {
return g.Id
}
func (g *RemoteGrain) GetCurrentState() ([]byte, error) {
err := SendCartPacket(g.client, g.Id, RemoteGetState, func(w io.Writer) error {
return nil
})
if err != nil {
return nil, err
}
_, data, err := ReceivePacket(g.client)
return data, err
}
func NewRemoteGrainPool(addr ...string) *RemoteGrainPool {
return &RemoteGrainPool{
Hosts: addr,
grains: make(map[CartId]RemoteGrain),
}
}
func (p *RemoteGrainPool) findRemoteGrain(id CartId) *RemoteGrain {
grain, ok := p.grains[id]
if !ok {
return nil
}
return &grain
}
func (p *RemoteGrainPool) Process(id CartId, messages ...Message) ([]byte, error) {
var result []byte
var err error
grain := p.findRemoteGrain(id)
if grain == nil {
grain = NewRemoteGrain(id, p.Hosts[0])
grain.Connect()
p.grains[id] = *grain
}
for _, message := range messages {
result, err = grain.HandleMessage(&message, false)
}
return result, err
}
func (p *RemoteGrainPool) Get(id CartId) ([]byte, error) {
grain := p.findRemoteGrain(id)
if grain == nil {
return nil, nil
}
return grain.GetCurrentState()
}

114
rpc-server.go Normal file
View File

@@ -0,0 +1,114 @@
package main
import (
"encoding/binary"
"encoding/json"
"fmt"
"io"
"net"
)
type GrainHandler struct {
listener net.Listener
pool GrainPool
}
func (h *GrainHandler) GetState(id CartId, reply *Grain) error {
grain, err := h.pool.Get(id)
if err != nil {
return err
}
*reply = grain
return nil
}
func NewGrainHandler(pool GrainPool, listen string) (*GrainHandler, error) {
handler := &GrainHandler{
pool: pool,
}
l, err := net.Listen("tcp", listen)
handler.listener = l
return handler, err
}
func (h *GrainHandler) Serve() {
for {
// Accept incoming connections
conn, err := h.listener.Accept()
if err != nil {
fmt.Println("Error:", err)
continue
}
// Handle client connection in a goroutine
go h.handleClient(conn)
}
}
func (h *GrainHandler) handleClient(conn net.Conn) {
fmt.Println("Handling client connection")
defer conn.Close()
var packet CartPacket
for {
for {
err := binary.Read(conn, binary.LittleEndian, &packet)
if err != nil {
if err == io.EOF {
break
}
fmt.Println("Error reading packet:", err)
}
if packet.Version != 2 {
fmt.Printf("Unknown version %d", packet.Version)
break
}
switch packet.MessageType {
case RemoteHandleMessage:
fmt.Printf("Handling message\n")
var msg Message
err := MessageFromReader(conn, &msg)
if err != nil {
fmt.Println("Error reading message:", err)
}
fmt.Printf("Message: %s, %v\n", packet.Id.String(), msg)
grain, err := h.pool.Get(packet.Id)
if err != nil {
fmt.Println("Error getting grain:", err)
}
_, err = grain.HandleMessage(&msg, false)
if err != nil {
fmt.Println("Error handling message:", err)
}
SendPacket(conn, ResponseBody, func(w io.Writer) error {
data, err := json.Marshal(grain)
if err != nil {
return err
}
w.Write(data)
return nil
})
case RemoteGetState:
fmt.Printf("Package: %s %v\n", packet.Id.String(), packet)
grain, err := h.pool.Get(packet.Id)
if err != nil {
fmt.Println("Error getting grain:", err)
}
SendPacket(conn, ResponseBody, func(w io.Writer) error {
data, err := json.Marshal(grain)
if err != nil {
return err
}
w.Write(data)
return nil
})
}
}
}
}