Scala / Akka / Protobuf: не удалось сериализовать и десериализовать сообщение - PullRequest
Новая
       39

Scala / Akka / Protobuf: не удалось сериализовать и десериализовать сообщение

0 голосов
/ 25 марта 2019

мы пытаемся использовать protobuf с Akka и сериализовать все сообщения Akka через protobuf.Для Scala у нас есть библиотека ScalaPB, которая помогает нам генерировать класс, который включает методы типа parseFrom, toByteArray и т. Д. Для сериализации или десериализации наших данных.Но, пока мы пытаемся запустить программу, получаем следующее исключение: 

akka.actor.dungeon.SerializationCheckFailedException: Failed to serialize and deserialize message of type com.knoldus.akkaserialization.msg.example.Bang$ for testing. To avoid this error, either disable 'akka.actor.serialize-messages', mark the message with 'akka.actor.NoSerializationVerificationNeeded', or configure serialization to support this message

Файл application.conf содержит следующую конфигурацию: 

akka {
    actor {
        allow-java-serialization = off
        serialize-messages = on
        serializers {
            proto = "akka.remote.serialization.ProtobufSerializer"
        }

        serialization-bindings {
            "com.knoldus.akkaserialization.msg.example.Bang" = proto
            "com.knoldus.akkaserialization.msg.example.Message" = proto
        }
    }
}

Эти классы com.knoldus.akkaserialization.msg.example.Bang и com.knoldus.akkaserialization.msg.example.Message генерируютчерез ScalaPB и содержит все необходимые методы.

Исходный код akka.remote.serialization.ProtobufSerializer define, 

This Serializer serializes `akka.protobuf.Message` and `com.google.protobuf.Message` It is using reflection to find the `parseFrom` and `toByteArray` methods to avoid dependency to `com.google.protobuf`

Итак, мы ожидаем, что это автоматически считывает наши классы дел Bang и Message и выполняет сериализацию, но, к сожалению, получает исключение сериализации,

Не могли бы вы выяснить, в чем конкретно проблема с ScalaPB и ProtoBuff?

1 Ответ

1 голос
/ 25 марта 2019

Сериализатор, который вы используете, был разработан для работы с протоколами Java, а не с ScalaPB.Вы должны включить свой собственный сериализатор.Вот один из них, который вы можете использовать: 

package com.example.protoser

import java.util.concurrent.atomic.AtomicReference

import akka.actor.ExtendedActorSystem
import akka.serialization.BaseSerializer
import scalapb.GeneratedMessageCompanion

class ScalaPbSerializer(val system: ExtendedActorSystem) extends BaseSerializer {
  private val classToCompanionMapRef = new AtomicReference[Map[Class[_], GeneratedMessageCompanion[_]]](Map.empty)

  override def toBinary(o: AnyRef): Array[Byte] = o match {
    case e: scalapb.GeneratedMessage => e.toByteArray
    case _ => throw new IllegalArgumentException("Need a subclass of scalapb.GeneratedMessage")
  }

  override def includeManifest: Boolean = true

  override def fromBinary(bytes: Array[Byte], manifest: Option[Class[_]]): AnyRef = {
    manifest match {
      case Some(clazz) =>
        @scala.annotation.tailrec
        def messageCompanion(companion: GeneratedMessageCompanion[_] = null): GeneratedMessageCompanion[_] = {
          val classToCompanion = classToCompanionMapRef.get()
          classToCompanion.get(clazz) match {
            case Some(cachedCompanion) => cachedCompanion
            case None =>
              val uncachedCompanion =
                if (companion eq null) Class.forName(clazz.getName + "$", true, clazz.getClassLoader)
                  .getField("MODULE$").get().asInstanceOf[GeneratedMessageCompanion[_]]
                else companion
              if (classToCompanionMapRef.compareAndSet(classToCompanion, classToCompanion.updated(clazz, uncachedCompanion)))
                uncachedCompanion
              else
                messageCompanion(uncachedCompanion)
          }
        }
        messageCompanion().parseFrom(bytes).asInstanceOf[AnyRef]
      case _ => throw new IllegalArgumentException("Need a ScalaPB companion class to be able to deserialize.")
    }
  }
}

Конфигурация должна быть примерно такой: 

akka {
  actor {
    serializers {
      scalapb = "com.example.protoser.ScalaPbSerializer"
    }

    serialization-bindings {
      "scalapb.GeneratedMessage" = scalapb
    }

    serialization-identifiers {
      "com.example.protoser.ScalaPbSerializer" = 10000
    }
  }
}

Выше был адаптер из старого кода, поэтому изменения и предложения приветствуются!

...