@ -47,21 +47,26 @@ func (s *AuthPostmanServer) ReadMessage(offset int64) error {
//
//
for {
for {
m , err := s . kafkaReader . ReadMessage ( )
// m, err := s.kafkaReader.ReadMessage()
m , err := s . kafkaReader . FetchMessage ( )
if err != nil {
if err != nil {
return err
s . logger . Error ( err )
continue
}
}
// Декодируем сообщение
// Декодируем сообщение
msg := api . MessageRegistration { }
msg := api . MessageRegistration { }
erk := json . Unmarshal ( m . Value , & msg )
erk := json . Unmarshal ( m . Value , & msg )
if erk != nil {
if erk != nil {
return erk
s . logger . Error ( erk )
continue
}
}
// Проверим пришедшее сообщение, не является ли оно повтором
// Проверим пришедшее сообщение, не является ли оно повтором
if _ , ok := s . msgs [ msg . ID ] ; ok {
if _ , ok := s . msgs [ msg . ID ] ; ok {
s . logger . Print f( "the message #%d is a duplicate" , msg . ID )
s . logger . Warn f( "the message #%d is a duplicate" , msg . ID )
continue
continue
} else {
} else {
@ -82,9 +87,17 @@ func (s *AuthPostmanServer) ReadMessage(offset int64) error {
message . AppendRecipient ( msg . Email )
message . AppendRecipient ( msg . Email )
ers := postman . Send ( message )
ers := postman . Send ( message )
if ers != nil {
if ers != nil {
s . logger . Print ( ers )
s . logger . Error ( ers )
continue
} else {
} else {
s . logger . Printf ( "send code %s to %s (message #%d) completed" , msg . Code , msg . Email , msg . ID )
s . logger . Printf ( "send code %s to %s (message #%d) completed" , msg . Code , msg . Email , msg . ID )
}
}
// Подтвердим сообщение после успешной обработки
erf := s . kafkaReader . CommitMessage ( m )
if erf != nil {
s . logger . Error ( erf )
}
}
}
}
}