+1 -1

example/main.go

··· 33 33 }() 34 34 startAt := 0 35 35 startAtType := server.Current 36 36 - if *consumeFrom >= 0-1 { 36 36 + if *consumeFrom > -1 { 37 37 startAtType = server.From 38 38 startAt = *consumeFrom 39 39 }

+3 -11

server/server.go

··· 75 75 From StartAtType = 2 76 76 ) 77 77 78 78 - type Store interface { 79 79 - Write(msg MessageToSend) error 80 80 - ReadFrom(offset int, handleFunc func(msg MessageToSend)) error 81 81 - } 82 82 - 83 78 // Server accepts subscribe and publish connections and passes messages around 84 79 type Server struct { 85 80 Addr string ··· 222 217 } 223 218 224 219 var topics []string 225 225 - fmt.Println(string(buf)) 226 220 err = json.Unmarshal(buf, &topics) 227 221 if err != nil { 228 222 slog.Error("failed to unmarshal subscibers topic data", "error", err, "peer", peer.Addr()) ··· 316 310 func (s *Server) handlePublish(peer *peer.Peer) { 317 311 slog.Info("handling publisher", "peer", peer.Addr()) 318 312 for { 319 319 - var message *MessageToSend 320 320 - 321 313 op := func(conn net.Conn) error { 322 314 dataLen, err := dataLength(conn) 323 315 if err != nil { ··· 365 357 return nil 366 358 } 367 359 368 368 - message = &MessageToSend{ 360 360 + message := MessageToSend{ 369 361 topic: topicStr, 370 362 data: dataBuf, 371 363 } ··· 376 368 s.topics[message.topic] = topic 377 369 } 378 370 379 379 - err = topic.sendMessageToSubscribers(*message) 371 371 + err = topic.sendMessageToSubscribers(message) 380 372 if err != nil { 381 373 slog.Error("failed to send message to subscribers", "error", err, "peer", peer.Addr()) 382 374 writeStatus(Error, "failed to send message to subscribers", conn) ··· 406 398 t = newTopic(topicName) 407 399 } 408 400 409 409 - t.subscriptions[peer.Addr()] = newSubscriber(peer, topicName, s.ackDelay, s.ackTimeout, t.messageStore, startAt) 401 401 + t.subscriptions[peer.Addr()] = newSubscriber(peer, t, s.ackDelay, s.ackTimeout, startAt) 410 402 411 403 s.topics[topicName] = t 412 404 }

+3 -3

server/subscriber.go

··· 29 29 return message{data: data, deliveryCount: 1} 30 30 } 31 31 32 32 - func newSubscriber(peer *peer.Peer, topic string, ackDelay, ackTimeout time.Duration, messageStore Store, startAt int) *subscriber { 32 32 + func newSubscriber(peer *peer.Peer, topic *topic, ackDelay, ackTimeout time.Duration, startAt int) *subscriber { 33 33 s := &subscriber{ 34 34 peer: peer, 35 35 - topic: topic, 35 35 + topic: topic.name, 36 36 messages: make(chan message), 37 37 ackDelay: ackDelay, 38 38 ackTimeout: ackTimeout, ··· 44 44 offset := startAt 45 45 46 46 go func() { 47 47 - err := messageStore.ReadFrom(offset, func(msg MessageToSend) { 47 47 + err := topic.messageStore.ReadFrom(offset, func(msg MessageToSend) { 48 48 s.messages <- newMessage(msg.data) 49 49 }) 50 50 if err != nil {

+5

server/topic.go

··· 6 6 "sync" 7 7 ) 8 8 9 9 + type Store interface { 10 10 + Write(msg MessageToSend) error 11 11 + ReadFrom(offset int, handleFunc func(msg MessageToSend)) error 12 12 + } 13 13 + 9 14 type topic struct { 10 15 name string 11 16 subscriptions map[net.Addr]*subscriber