shared: Split P2P Topics And Introduce Middleware (Adapters) (#421)

2026-01-08 23:18:15 -05:00 · 2018-08-29 12:32:54 -04:00
parent 5adc94b2ae
commit b02042dbe9
27 changed files with 869 additions and 509 deletions
--- a/6
+++ b/6
@@ -566,3 +566,9 @@ go_repository(
    commit = "c4c61651e9e37fa117f53c5a906d3b63090d8445",
    importpath = "github.com/syndtr/goleveldb",
 )
+
+go_repository(
+    name = "com_github_libp2p_go_libp2p_blankhost",
+    commit = "073f507db72de824e981aa0f15f158175a8d6be1",
+    importpath = "github.com/libp2p/go-libp2p-blankhost",
+)
--- a/beacon-chain/node/BUILD.bazel
+++ b/beacon-chain/node/BUILD.bazel
@@ -2,7 +2,10 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")

 go_library(
    name = "go_default_library",
-    srcs = ["node.go"],
+    srcs = [
+        "node.go",
+        "p2p_config.go",
+    ],
    importpath = "github.com/prysmaticlabs/prysm/beacon-chain/node",
    visibility = ["//beacon-chain:__subpackages__"],
    deps = [
@@ -13,6 +16,7 @@ go_library(
        "//beacon-chain/sync:go_default_library",
        "//beacon-chain/sync/initial-sync:go_default_library",
        "//beacon-chain/utils:go_default_library",
+        "//proto/beacon/p2p/v1:go_default_library",
        "//shared:go_default_library",
        "//shared/cmd:go_default_library",
        "//shared/database:go_default_library",
--- a/beacon-chain/node/node.go
+++ b/beacon-chain/node/node.go
@@ -144,10 +144,11 @@ func (b *BeaconNode) startDB(ctx *cli.Context) error {
 }

 func (b *BeaconNode) registerP2P() error {
-	beaconp2p, err := p2p.NewServer()
+	beaconp2p, err := configureP2P()
 	if err != nil {
 		return fmt.Errorf("could not register p2p service: %v", err)
 	}
+
 	return b.services.RegisterService(beaconp2p)
 }

--- a/beacon-chain/node/p2p_config.go
+++ b/beacon-chain/node/p2p_config.go
@@ -0,0 +1,35 @@
+package node
+
+import (
+	"github.com/prysmaticlabs/prysm/shared/p2p"
+
+	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
+)
+
+var topicMappings = map[pb.Topic]interface{}{
+	pb.Topic_BEACON_BLOCK_HASH_ANNOUNCE:          pb.BeaconBlockHashAnnounce{},
+	pb.Topic_BEACON_BLOCK_REQUEST:                pb.BeaconBlockRequest{},
+	pb.Topic_BEACON_BLOCK_REQUEST_BY_SLOT_NUMBER: pb.BeaconBlockRequestBySlotNumber{},
+	pb.Topic_BEACON_BLOCK_RESPONSE:               pb.BeaconBlockResponse{},
+	pb.Topic_CRYSTALLIZED_STATE_HASH_ANNOUNCE:    pb.CrystallizedStateHashAnnounce{},
+	pb.Topic_CRYSTALLIZED_STATE_REQUEST:          pb.CrystallizedStateRequest{},
+	pb.Topic_CRYSTALLIZED_STATE_RESPONSE:         pb.CrystallizedStateResponse{},
+	pb.Topic_ACTIVE_STATE_HASH_ANNOUNCE:          pb.ActiveStateHashAnnounce{},
+	pb.Topic_ACTIVE_STATE_REQUEST:                pb.ActiveStateRequest{},
+	pb.Topic_ACTIVE_STATE_RESPONSE:               pb.ActiveStateResponse{},
+}
+
+func configureP2P() (*p2p.Server, error) {
+	s, err := p2p.NewServer()
+	if err != nil {
+		return nil, err
+	}
+
+	// TODO(437, 438): Define default adapters for logging, monitoring, etc.
+	var adapters []p2p.Adapter
+	for k, v := range topicMappings {
+		s.RegisterTopic(k.String(), v, adapters...)
+	}
+
+	return s, nil
+}
--- a/beacon-chain/sync/BUILD.bazel
+++ b/beacon-chain/sync/BUILD.bazel
@@ -18,6 +18,7 @@ go_test(
    name = "go_default_test",
    srcs = ["service_test.go"],
    embed = [":go_default_library"],
+    race = "off",  # TODO(#377): fix issues with race detection testing.
    deps = [
        "//proto/beacon/p2p/v1:go_default_library",
        "//shared/p2p:go_default_library",
--- a/proto/README.md
+++ b/proto/README.md
@@ -12,6 +12,18 @@ proto/
  sharding/
    p2p/
      v1/
+  testing/
 ```

-We specify messages available for p2p communication common to beacon chain nodes and sharding clients.
+We specify messages available for p2p communication common to beacon chain nodes and sharding clients.
+
+For now, we are checking in all generated code to support native go dependency
+management. The generated pb.go files can be derived from bazel's bin 
+directory.
+
+For example, when we build the testing go proto library 
+`bazel build //proto/testing:ethereum_testing_go_proto` there is a pb.go 
+generated at 
+`bazel-bin/proto/testing/linux_amd64_stripped/ethereum_testing_go_proto\~/github.com/prysmaticlabs/prysm/proto/testing/test.pb.go`.
+This generated file can be copied, or you can use you protoc locally if you
+prefer.
--- a/proto/beacon/p2p/v1/messages.pb.go
+++ b/proto/beacon/p2p/v1/messages.pb.go
@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // source: proto/beacon/p2p/v1/messages.proto

-package ethereum_beacon_p2p_v1
+package v1

 import proto "github.com/golang/protobuf/proto"
 import fmt "fmt"
@@ -19,6 +19,56 @@ var _ = math.Inf
 // proto package needs to be updated.
 const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package

+type Topic int32
+
+const (
+	Topic_UNKNOWN                             Topic = 0
+	Topic_BEACON_BLOCK_HASH_ANNOUNCE          Topic = 1
+	Topic_BEACON_BLOCK_REQUEST                Topic = 2
+	Topic_BEACON_BLOCK_REQUEST_BY_SLOT_NUMBER Topic = 3
+	Topic_BEACON_BLOCK_RESPONSE               Topic = 4
+	Topic_CRYSTALLIZED_STATE_HASH_ANNOUNCE    Topic = 5
+	Topic_CRYSTALLIZED_STATE_REQUEST          Topic = 6
+	Topic_CRYSTALLIZED_STATE_RESPONSE         Topic = 7
+	Topic_ACTIVE_STATE_HASH_ANNOUNCE          Topic = 8
+	Topic_ACTIVE_STATE_REQUEST                Topic = 9
+	Topic_ACTIVE_STATE_RESPONSE               Topic = 10
+)
+
+var Topic_name = map[int32]string{
+	0:  "UNKNOWN",
+	1:  "BEACON_BLOCK_HASH_ANNOUNCE",
+	2:  "BEACON_BLOCK_REQUEST",
+	3:  "BEACON_BLOCK_REQUEST_BY_SLOT_NUMBER",
+	4:  "BEACON_BLOCK_RESPONSE",
+	5:  "CRYSTALLIZED_STATE_HASH_ANNOUNCE",
+	6:  "CRYSTALLIZED_STATE_REQUEST",
+	7:  "CRYSTALLIZED_STATE_RESPONSE",
+	8:  "ACTIVE_STATE_HASH_ANNOUNCE",
+	9:  "ACTIVE_STATE_REQUEST",
+	10: "ACTIVE_STATE_RESPONSE",
+}
+var Topic_value = map[string]int32{
+	"UNKNOWN":                             0,
+	"BEACON_BLOCK_HASH_ANNOUNCE":          1,
+	"BEACON_BLOCK_REQUEST":                2,
+	"BEACON_BLOCK_REQUEST_BY_SLOT_NUMBER": 3,
+	"BEACON_BLOCK_RESPONSE":               4,
+	"CRYSTALLIZED_STATE_HASH_ANNOUNCE":    5,
+	"CRYSTALLIZED_STATE_REQUEST":          6,
+	"CRYSTALLIZED_STATE_RESPONSE":         7,
+	"ACTIVE_STATE_HASH_ANNOUNCE":          8,
+	"ACTIVE_STATE_REQUEST":                9,
+	"ACTIVE_STATE_RESPONSE":               10,
+}
+
+func (x Topic) String() string {
+	return proto.EnumName(Topic_name, int32(x))
+}
+func (Topic) EnumDescriptor() ([]byte, []int) {
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{0}
+}
+
 type BeaconBlockHashAnnounce struct {
 	Hash                 []byte   `protobuf:"bytes,1,opt,name=hash,proto3" json:"hash,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
@@ -30,7 +80,7 @@ func (m *BeaconBlockHashAnnounce) Reset()         { *m = BeaconBlockHashAnnounce
 func (m *BeaconBlockHashAnnounce) String() string { return proto.CompactTextString(m) }
 func (*BeaconBlockHashAnnounce) ProtoMessage()    {}
 func (*BeaconBlockHashAnnounce) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{0}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{0}
 }
 func (m *BeaconBlockHashAnnounce) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_BeaconBlockHashAnnounce.Unmarshal(m, b)
@@ -68,7 +118,7 @@ func (m *BeaconBlockRequest) Reset()         { *m = BeaconBlockRequest{} }
 func (m *BeaconBlockRequest) String() string { return proto.CompactTextString(m) }
 func (*BeaconBlockRequest) ProtoMessage()    {}
 func (*BeaconBlockRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{1}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{1}
 }
 func (m *BeaconBlockRequest) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_BeaconBlockRequest.Unmarshal(m, b)
@@ -96,7 +146,7 @@ func (m *BeaconBlockRequest) GetHash() []byte {
 }

 type BeaconBlockRequestBySlotNumber struct {
-	SlotNumber           uint64   `protobuf:"varint,1,opt,name=slot_number,json=slotNumber,proto3" json:"slot_number,omitempty"`
+	SlotNumber           uint64   `protobuf:"varint,1,opt,name=slot_number,json=slotNumber" json:"slot_number,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -106,7 +156,7 @@ func (m *BeaconBlockRequestBySlotNumber) Reset()         { *m = BeaconBlockReque
 func (m *BeaconBlockRequestBySlotNumber) String() string { return proto.CompactTextString(m) }
 func (*BeaconBlockRequestBySlotNumber) ProtoMessage()    {}
 func (*BeaconBlockRequestBySlotNumber) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{2}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{2}
 }
 func (m *BeaconBlockRequestBySlotNumber) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_BeaconBlockRequestBySlotNumber.Unmarshal(m, b)
@@ -134,7 +184,7 @@ func (m *BeaconBlockRequestBySlotNumber) GetSlotNumber() uint64 {
 }

 type BeaconBlockResponse struct {
-	Block                *BeaconBlock `protobuf:"bytes,1,opt,name=block,proto3" json:"block,omitempty"`
+	Block                *BeaconBlock `protobuf:"bytes,1,opt,name=block" json:"block,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}     `json:"-"`
 	XXX_unrecognized     []byte       `json:"-"`
 	XXX_sizecache        int32        `json:"-"`
@@ -144,7 +194,7 @@ func (m *BeaconBlockResponse) Reset()         { *m = BeaconBlockResponse{} }
 func (m *BeaconBlockResponse) String() string { return proto.CompactTextString(m) }
 func (*BeaconBlockResponse) ProtoMessage()    {}
 func (*BeaconBlockResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{3}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{3}
 }
 func (m *BeaconBlockResponse) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_BeaconBlockResponse.Unmarshal(m, b)
@@ -173,13 +223,13 @@ func (m *BeaconBlockResponse) GetBlock() *BeaconBlock {

 type BeaconBlock struct {
 	ParentHash            []byte               `protobuf:"bytes,1,opt,name=parent_hash,json=parentHash,proto3" json:"parent_hash,omitempty"`
-	SlotNumber            uint64               `protobuf:"varint,2,opt,name=slot_number,json=slotNumber,proto3" json:"slot_number,omitempty"`
+	SlotNumber            uint64               `protobuf:"varint,2,opt,name=slot_number,json=slotNumber" json:"slot_number,omitempty"`
 	RandaoReveal          []byte               `protobuf:"bytes,3,opt,name=randao_reveal,json=randaoReveal,proto3" json:"randao_reveal,omitempty"`
 	PowChainRef           []byte               `protobuf:"bytes,4,opt,name=pow_chain_ref,json=powChainRef,proto3" json:"pow_chain_ref,omitempty"`
 	ActiveStateHash       []byte               `protobuf:"bytes,5,opt,name=active_state_hash,json=activeStateHash,proto3" json:"active_state_hash,omitempty"`
 	CrystallizedStateHash []byte               `protobuf:"bytes,6,opt,name=crystallized_state_hash,json=crystallizedStateHash,proto3" json:"crystallized_state_hash,omitempty"`
-	Timestamp             *timestamp.Timestamp `protobuf:"bytes,7,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
-	Attestations          []*AttestationRecord `protobuf:"bytes,8,rep,name=attestations,proto3" json:"attestations,omitempty"`
+	Timestamp             *timestamp.Timestamp `protobuf:"bytes,7,opt,name=timestamp" json:"timestamp,omitempty"`
+	Attestations          []*AttestationRecord `protobuf:"bytes,8,rep,name=attestations" json:"attestations,omitempty"`
 	XXX_NoUnkeyedLiteral  struct{}             `json:"-"`
 	XXX_unrecognized      []byte               `json:"-"`
 	XXX_sizecache         int32                `json:"-"`
@@ -189,7 +239,7 @@ func (m *BeaconBlock) Reset()         { *m = BeaconBlock{} }
 func (m *BeaconBlock) String() string { return proto.CompactTextString(m) }
 func (*BeaconBlock) ProtoMessage()    {}
 func (*BeaconBlock) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{4}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{4}
 }
 func (m *BeaconBlock) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_BeaconBlock.Unmarshal(m, b)
@@ -276,7 +326,7 @@ func (m *CrystallizedStateHashAnnounce) Reset()         { *m = CrystallizedState
 func (m *CrystallizedStateHashAnnounce) String() string { return proto.CompactTextString(m) }
 func (*CrystallizedStateHashAnnounce) ProtoMessage()    {}
 func (*CrystallizedStateHashAnnounce) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{5}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{5}
 }
 func (m *CrystallizedStateHashAnnounce) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_CrystallizedStateHashAnnounce.Unmarshal(m, b)
@@ -314,7 +364,7 @@ func (m *CrystallizedStateRequest) Reset()         { *m = CrystallizedStateReque
 func (m *CrystallizedStateRequest) String() string { return proto.CompactTextString(m) }
 func (*CrystallizedStateRequest) ProtoMessage()    {}
 func (*CrystallizedStateRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{6}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{6}
 }
 func (m *CrystallizedStateRequest) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_CrystallizedStateRequest.Unmarshal(m, b)
@@ -342,7 +392,7 @@ func (m *CrystallizedStateRequest) GetHash() []byte {
 }

 type CrystallizedStateResponse struct {
-	CrystallizedState    *CrystallizedState `protobuf:"bytes,1,opt,name=crystallized_state,json=crystallizedState,proto3" json:"crystallized_state,omitempty"`
+	CrystallizedState    *CrystallizedState `protobuf:"bytes,1,opt,name=crystallized_state,json=crystallizedState" json:"crystallized_state,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}           `json:"-"`
 	XXX_unrecognized     []byte             `json:"-"`
 	XXX_sizecache        int32              `json:"-"`
@@ -352,7 +402,7 @@ func (m *CrystallizedStateResponse) Reset()         { *m = CrystallizedStateResp
 func (m *CrystallizedStateResponse) String() string { return proto.CompactTextString(m) }
 func (*CrystallizedStateResponse) ProtoMessage()    {}
 func (*CrystallizedStateResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{7}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{7}
 }
 func (m *CrystallizedStateResponse) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_CrystallizedStateResponse.Unmarshal(m, b)
@@ -380,18 +430,18 @@ func (m *CrystallizedStateResponse) GetCrystallizedState() *CrystallizedState {
 }

 type CrystallizedState struct {
-	LastStateRecalc        uint64                    `protobuf:"varint,1,opt,name=last_state_recalc,json=lastStateRecalc,proto3" json:"last_state_recalc,omitempty"`
-	JustifiedStreak        uint64                    `protobuf:"varint,2,opt,name=justified_streak,json=justifiedStreak,proto3" json:"justified_streak,omitempty"`
-	LastJustifiedSlot      uint64                    `protobuf:"varint,3,opt,name=last_justified_slot,json=lastJustifiedSlot,proto3" json:"last_justified_slot,omitempty"`
-	LastFinalizedSlot      uint64                    `protobuf:"varint,4,opt,name=last_finalized_slot,json=lastFinalizedSlot,proto3" json:"last_finalized_slot,omitempty"`
-	CurrentDynasty         uint64                    `protobuf:"varint,5,opt,name=current_dynasty,json=currentDynasty,proto3" json:"current_dynasty,omitempty"`
-	CrosslinkingStartShard uint64                    `protobuf:"varint,6,opt,name=crosslinking_start_shard,json=crosslinkingStartShard,proto3" json:"crosslinking_start_shard,omitempty"`
-	TotalDeposits          uint64                    `protobuf:"varint,7,opt,name=total_deposits,json=totalDeposits,proto3" json:"total_deposits,omitempty"`
+	LastStateRecalc        uint64                    `protobuf:"varint,1,opt,name=last_state_recalc,json=lastStateRecalc" json:"last_state_recalc,omitempty"`
+	JustifiedStreak        uint64                    `protobuf:"varint,2,opt,name=justified_streak,json=justifiedStreak" json:"justified_streak,omitempty"`
+	LastJustifiedSlot      uint64                    `protobuf:"varint,3,opt,name=last_justified_slot,json=lastJustifiedSlot" json:"last_justified_slot,omitempty"`
+	LastFinalizedSlot      uint64                    `protobuf:"varint,4,opt,name=last_finalized_slot,json=lastFinalizedSlot" json:"last_finalized_slot,omitempty"`
+	CurrentDynasty         uint64                    `protobuf:"varint,5,opt,name=current_dynasty,json=currentDynasty" json:"current_dynasty,omitempty"`
+	CrosslinkingStartShard uint64                    `protobuf:"varint,6,opt,name=crosslinking_start_shard,json=crosslinkingStartShard" json:"crosslinking_start_shard,omitempty"`
+	TotalDeposits          uint64                    `protobuf:"varint,7,opt,name=total_deposits,json=totalDeposits" json:"total_deposits,omitempty"`
 	DynastySeed            []byte                    `protobuf:"bytes,8,opt,name=dynasty_seed,json=dynastySeed,proto3" json:"dynasty_seed,omitempty"`
-	DynastySeedLastReset   uint64                    `protobuf:"varint,9,opt,name=dynasty_seed_last_reset,json=dynastySeedLastReset,proto3" json:"dynasty_seed_last_reset,omitempty"`
-	CrosslinkRecords       []*CrosslinkRecord        `protobuf:"bytes,10,rep,name=crosslink_records,json=crosslinkRecords,proto3" json:"crosslink_records,omitempty"`
-	Validators             []*ValidatorRecord        `protobuf:"bytes,11,rep,name=validators,proto3" json:"validators,omitempty"`
-	IndicesForSlots        []*ShardAndCommitteeArray `protobuf:"bytes,12,rep,name=indices_for_slots,json=indicesForSlots,proto3" json:"indices_for_slots,omitempty"`
+	DynastySeedLastReset   uint64                    `protobuf:"varint,9,opt,name=dynasty_seed_last_reset,json=dynastySeedLastReset" json:"dynasty_seed_last_reset,omitempty"`
+	CrosslinkRecords       []*CrosslinkRecord        `protobuf:"bytes,10,rep,name=crosslink_records,json=crosslinkRecords" json:"crosslink_records,omitempty"`
+	Validators             []*ValidatorRecord        `protobuf:"bytes,11,rep,name=validators" json:"validators,omitempty"`
+	IndicesForSlots        []*ShardAndCommitteeArray `protobuf:"bytes,12,rep,name=indices_for_slots,json=indicesForSlots" json:"indices_for_slots,omitempty"`
 	XXX_NoUnkeyedLiteral   struct{}                  `json:"-"`
 	XXX_unrecognized       []byte                    `json:"-"`
 	XXX_sizecache          int32                     `json:"-"`
@@ -401,7 +451,7 @@ func (m *CrystallizedState) Reset()         { *m = CrystallizedState{} }
 func (m *CrystallizedState) String() string { return proto.CompactTextString(m) }
 func (*CrystallizedState) ProtoMessage()    {}
 func (*CrystallizedState) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{8}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{8}
 }
 func (m *CrystallizedState) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_CrystallizedState.Unmarshal(m, b)
@@ -506,7 +556,7 @@ func (m *CrystallizedState) GetIndicesForSlots() []*ShardAndCommitteeArray {
 }

 type ShardAndCommitteeArray struct {
-	ArrayShardAndCommittee []*ShardAndCommittee `protobuf:"bytes,1,rep,name=array_shard_and_committee,json=arrayShardAndCommittee,proto3" json:"array_shard_and_committee,omitempty"`
+	ArrayShardAndCommittee []*ShardAndCommittee `protobuf:"bytes,1,rep,name=array_shard_and_committee,json=arrayShardAndCommittee" json:"array_shard_and_committee,omitempty"`
 	XXX_NoUnkeyedLiteral   struct{}             `json:"-"`
 	XXX_unrecognized       []byte               `json:"-"`
 	XXX_sizecache          int32                `json:"-"`
@@ -516,7 +566,7 @@ func (m *ShardAndCommitteeArray) Reset()         { *m = ShardAndCommitteeArray{}
 func (m *ShardAndCommitteeArray) String() string { return proto.CompactTextString(m) }
 func (*ShardAndCommitteeArray) ProtoMessage()    {}
 func (*ShardAndCommitteeArray) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{9}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{9}
 }
 func (m *ShardAndCommitteeArray) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ShardAndCommitteeArray.Unmarshal(m, b)
@@ -554,7 +604,7 @@ func (m *ActiveStateHashAnnounce) Reset()         { *m = ActiveStateHashAnnounce
 func (m *ActiveStateHashAnnounce) String() string { return proto.CompactTextString(m) }
 func (*ActiveStateHashAnnounce) ProtoMessage()    {}
 func (*ActiveStateHashAnnounce) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{10}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{10}
 }
 func (m *ActiveStateHashAnnounce) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ActiveStateHashAnnounce.Unmarshal(m, b)
@@ -592,7 +642,7 @@ func (m *ActiveStateRequest) Reset()         { *m = ActiveStateRequest{} }
 func (m *ActiveStateRequest) String() string { return proto.CompactTextString(m) }
 func (*ActiveStateRequest) ProtoMessage()    {}
 func (*ActiveStateRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{11}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{11}
 }
 func (m *ActiveStateRequest) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ActiveStateRequest.Unmarshal(m, b)
@@ -620,8 +670,8 @@ func (m *ActiveStateRequest) GetHash() []byte {
 }

 type ShardAndCommittee struct {
-	ShardId              uint64   `protobuf:"varint,1,opt,name=shard_id,json=shardId,proto3" json:"shard_id,omitempty"`
-	Committee            []uint32 `protobuf:"varint,2,rep,packed,name=committee,proto3" json:"committee,omitempty"`
+	ShardId              uint64   `protobuf:"varint,1,opt,name=shard_id,json=shardId" json:"shard_id,omitempty"`
+	Committee            []uint32 `protobuf:"varint,2,rep,packed,name=committee" json:"committee,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -631,7 +681,7 @@ func (m *ShardAndCommittee) Reset()         { *m = ShardAndCommittee{} }
 func (m *ShardAndCommittee) String() string { return proto.CompactTextString(m) }
 func (*ShardAndCommittee) ProtoMessage()    {}
 func (*ShardAndCommittee) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{12}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{12}
 }
 func (m *ShardAndCommittee) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ShardAndCommittee.Unmarshal(m, b)
@@ -666,7 +716,7 @@ func (m *ShardAndCommittee) GetCommittee() []uint32 {
 }

 type ActiveStateResponse struct {
-	ActiveState          *ActiveState `protobuf:"bytes,1,opt,name=active_state,json=activeState,proto3" json:"active_state,omitempty"`
+	ActiveState          *ActiveState `protobuf:"bytes,1,opt,name=active_state,json=activeState" json:"active_state,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}     `json:"-"`
 	XXX_unrecognized     []byte       `json:"-"`
 	XXX_sizecache        int32        `json:"-"`
@@ -676,7 +726,7 @@ func (m *ActiveStateResponse) Reset()         { *m = ActiveStateResponse{} }
 func (m *ActiveStateResponse) String() string { return proto.CompactTextString(m) }
 func (*ActiveStateResponse) ProtoMessage()    {}
 func (*ActiveStateResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{13}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{13}
 }
 func (m *ActiveStateResponse) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ActiveStateResponse.Unmarshal(m, b)
@@ -704,7 +754,7 @@ func (m *ActiveStateResponse) GetActiveState() *ActiveState {
 }

 type ActiveState struct {
-	PendingAttestations  []*AttestationRecord `protobuf:"bytes,1,rep,name=pending_attestations,json=pendingAttestations,proto3" json:"pending_attestations,omitempty"`
+	PendingAttestations  []*AttestationRecord `protobuf:"bytes,1,rep,name=pending_attestations,json=pendingAttestations" json:"pending_attestations,omitempty"`
 	RecentBlockHashes    [][]byte             `protobuf:"bytes,2,rep,name=recent_block_hashes,json=recentBlockHashes,proto3" json:"recent_block_hashes,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}             `json:"-"`
 	XXX_unrecognized     []byte               `json:"-"`
@@ -715,7 +765,7 @@ func (m *ActiveState) Reset()         { *m = ActiveState{} }
 func (m *ActiveState) String() string { return proto.CompactTextString(m) }
 func (*ActiveState) ProtoMessage()    {}
 func (*ActiveState) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{14}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{14}
 }
 func (m *ActiveState) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ActiveState.Unmarshal(m, b)
@@ -750,13 +800,13 @@ func (m *ActiveState) GetRecentBlockHashes() [][]byte {
 }

 type ValidatorRecord struct {
-	PublicKey            uint64   `protobuf:"varint,1,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"`
-	WithdrawalShard      uint64   `protobuf:"varint,2,opt,name=withdrawal_shard,json=withdrawalShard,proto3" json:"withdrawal_shard,omitempty"`
+	PublicKey            uint64   `protobuf:"varint,1,opt,name=public_key,json=publicKey" json:"public_key,omitempty"`
+	WithdrawalShard      uint64   `protobuf:"varint,2,opt,name=withdrawal_shard,json=withdrawalShard" json:"withdrawal_shard,omitempty"`
 	WithdrawalAddress    []byte   `protobuf:"bytes,3,opt,name=withdrawal_address,json=withdrawalAddress,proto3" json:"withdrawal_address,omitempty"`
 	RandaoCommitment     []byte   `protobuf:"bytes,4,opt,name=randao_commitment,json=randaoCommitment,proto3" json:"randao_commitment,omitempty"`
-	Balance              uint64   `protobuf:"varint,5,opt,name=balance,proto3" json:"balance,omitempty"`
-	StartDynasty         uint64   `protobuf:"varint,6,opt,name=start_dynasty,json=startDynasty,proto3" json:"start_dynasty,omitempty"`
-	EndDynasty           uint64   `protobuf:"varint,7,opt,name=end_dynasty,json=endDynasty,proto3" json:"end_dynasty,omitempty"`
+	Balance              uint64   `protobuf:"varint,5,opt,name=balance" json:"balance,omitempty"`
+	StartDynasty         uint64   `protobuf:"varint,6,opt,name=start_dynasty,json=startDynasty" json:"start_dynasty,omitempty"`
+	EndDynasty           uint64   `protobuf:"varint,7,opt,name=end_dynasty,json=endDynasty" json:"end_dynasty,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -766,7 +816,7 @@ func (m *ValidatorRecord) Reset()         { *m = ValidatorRecord{} }
 func (m *ValidatorRecord) String() string { return proto.CompactTextString(m) }
 func (*ValidatorRecord) ProtoMessage()    {}
 func (*ValidatorRecord) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{15}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{15}
 }
 func (m *ValidatorRecord) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_ValidatorRecord.Unmarshal(m, b)
@@ -836,12 +886,12 @@ func (m *ValidatorRecord) GetEndDynasty() uint64 {
 }

 type AttestationRecord struct {
-	Slot                 uint64   `protobuf:"varint,1,opt,name=slot,proto3" json:"slot,omitempty"`
-	ShardId              uint64   `protobuf:"varint,2,opt,name=shard_id,json=shardId,proto3" json:"shard_id,omitempty"`
+	Slot                 uint64   `protobuf:"varint,1,opt,name=slot" json:"slot,omitempty"`
+	ShardId              uint64   `protobuf:"varint,2,opt,name=shard_id,json=shardId" json:"shard_id,omitempty"`
 	ShardBlockHash       []byte   `protobuf:"bytes,3,opt,name=shard_block_hash,json=shardBlockHash,proto3" json:"shard_block_hash,omitempty"`
 	AttesterBitfield     []byte   `protobuf:"bytes,4,opt,name=attester_bitfield,json=attesterBitfield,proto3" json:"attester_bitfield,omitempty"`
 	ObliqueParentHashes  [][]byte `protobuf:"bytes,5,rep,name=oblique_parent_hashes,json=obliqueParentHashes,proto3" json:"oblique_parent_hashes,omitempty"`
-	AggregateSig         []uint64 `protobuf:"varint,6,rep,packed,name=aggregate_sig,json=aggregateSig,proto3" json:"aggregate_sig,omitempty"`
+	AggregateSig         []uint64 `protobuf:"varint,6,rep,packed,name=aggregate_sig,json=aggregateSig" json:"aggregate_sig,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -851,7 +901,7 @@ func (m *AttestationRecord) Reset()         { *m = AttestationRecord{} }
 func (m *AttestationRecord) String() string { return proto.CompactTextString(m) }
 func (*AttestationRecord) ProtoMessage()    {}
 func (*AttestationRecord) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{16}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{16}
 }
 func (m *AttestationRecord) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_AttestationRecord.Unmarshal(m, b)
@@ -914,7 +964,7 @@ func (m *AttestationRecord) GetAggregateSig() []uint64 {
 }

 type CrosslinkRecord struct {
-	Dynasty              uint64   `protobuf:"varint,1,opt,name=dynasty,proto3" json:"dynasty,omitempty"`
+	Dynasty              uint64   `protobuf:"varint,1,opt,name=dynasty" json:"dynasty,omitempty"`
 	Blockhash            []byte   `protobuf:"bytes,2,opt,name=blockhash,proto3" json:"blockhash,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
@@ -925,7 +975,7 @@ func (m *CrosslinkRecord) Reset()         { *m = CrosslinkRecord{} }
 func (m *CrosslinkRecord) String() string { return proto.CompactTextString(m) }
 func (*CrosslinkRecord) ProtoMessage()    {}
 func (*CrosslinkRecord) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_a20b723c3c4695b7, []int{17}
+	return fileDescriptor_messages_bd1ed2903542c11b, []int{17}
 }
 func (m *CrosslinkRecord) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_CrosslinkRecord.Unmarshal(m, b)
@@ -978,83 +1028,94 @@ func init() {
 	proto.RegisterType((*ValidatorRecord)(nil), "ethereum.beacon.p2p.v1.ValidatorRecord")
 	proto.RegisterType((*AttestationRecord)(nil), "ethereum.beacon.p2p.v1.AttestationRecord")
 	proto.RegisterType((*CrosslinkRecord)(nil), "ethereum.beacon.p2p.v1.CrosslinkRecord")
+	proto.RegisterEnum("ethereum.beacon.p2p.v1.Topic", Topic_name, Topic_value)
 }

 func init() {
-	proto.RegisterFile("proto/beacon/p2p/v1/messages.proto", fileDescriptor_messages_a20b723c3c4695b7)
+	proto.RegisterFile("proto/beacon/p2p/v1/messages.proto", fileDescriptor_messages_bd1ed2903542c11b)
 }

-var fileDescriptor_messages_a20b723c3c4695b7 = []byte{
-	// 1121 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x56, 0xdd, 0x6e, 0x1b, 0x45,
-	0x14, 0x96, 0x13, 0xa7, 0x49, 0x8e, 0x9d, 0x38, 0x9e, 0xb4, 0xc9, 0xa6, 0xa2, 0x24, 0x6c, 0x85,
-	0x9a, 0x82, 0xba, 0x56, 0x53, 0x81, 0xca, 0xa5, 0x93, 0x2a, 0x50, 0x28, 0xa8, 0x5a, 0x57, 0x08,
-	0x21, 0xd0, 0x6a, 0xbc, 0x73, 0x6c, 0x0f, 0x59, 0xef, 0x6c, 0x67, 0xc6, 0x89, 0xcc, 0x05, 0x77,
-	0x3c, 0x03, 0x17, 0x3c, 0x01, 0xaf, 0xc7, 0x13, 0xa0, 0xf9, 0xf1, 0x7a, 0x1d, 0x27, 0xa9, 0xb8,
-	0xdb, 0xfd, 0xce, 0x77, 0xce, 0x9e, 0x9f, 0x6f, 0xce, 0x2c, 0x84, 0x85, 0x14, 0x5a, 0x74, 0xfa,
-	0x48, 0x53, 0x91, 0x77, 0x8a, 0x93, 0xa2, 0x73, 0xf9, 0xbc, 0x33, 0x46, 0xa5, 0xe8, 0x10, 0x55,
-	0x64, 0x8d, 0x64, 0x0f, 0xf5, 0x08, 0x25, 0x4e, 0xc6, 0x91, 0xa3, 0x45, 0xc5, 0x49, 0x11, 0x5d,
-	0x3e, 0x7f, 0x78, 0x38, 0x14, 0x62, 0x98, 0x61, 0xc7, 0xb2, 0xfa, 0x93, 0x41, 0x47, 0xf3, 0x31,
-	0x2a, 0x4d, 0xc7, 0x85, 0x73, 0x0c, 0x9f, 0xc1, 0xfe, 0xa9, 0xf5, 0x38, 0xcd, 0x44, 0x7a, 0xf1,
-	0x0d, 0x55, 0xa3, 0x6e, 0x9e, 0x8b, 0x49, 0x9e, 0x22, 0x21, 0x50, 0x1f, 0x51, 0x35, 0x0a, 0x6a,
-	0x47, 0xb5, 0xe3, 0x66, 0x6c, 0x9f, 0xc3, 0x63, 0x20, 0x15, 0x7a, 0x8c, 0xef, 0x27, 0xa8, 0xf4,
-	0x8d, 0xcc, 0x2e, 0x7c, 0xbc, 0xcc, 0x3c, 0x9d, 0xf6, 0x32, 0xa1, 0x7f, 0x98, 0x8c, 0xfb, 0x28,
-	0xc9, 0x21, 0x34, 0x54, 0x26, 0x74, 0x92, 0xdb, 0x57, 0xeb, 0x5c, 0x8f, 0x41, 0x95, 0x84, 0xf0,
-	0x2d, 0xec, 0x2e, 0x84, 0x50, 0x85, 0xc8, 0x15, 0x92, 0xaf, 0x60, 0xad, 0x6f, 0x00, 0xeb, 0xd1,
-	0x38, 0x79, 0x1c, 0xdd, 0x5c, 0x7b, 0x54, 0xf5, 0x75, 0x1e, 0xe1, 0x9f, 0xab, 0xd0, 0xa8, 0xc0,
-	0x26, 0x85, 0x82, 0x4a, 0xcc, 0x75, 0x52, 0xc9, 0x1f, 0x1c, 0x64, 0x7a, 0x71, 0x3d, 0xc7, 0x95,
-	0xeb, 0x39, 0x92, 0xc7, 0xb0, 0x25, 0x69, 0xce, 0xa8, 0x48, 0x24, 0x5e, 0x22, 0xcd, 0x82, 0x55,
-	0x1b, 0xa3, 0xe9, 0xc0, 0xd8, 0x62, 0x24, 0x84, 0xad, 0x42, 0x5c, 0x25, 0xe9, 0x88, 0xf2, 0x3c,
-	0x91, 0x38, 0x08, 0xea, 0x96, 0xd4, 0x28, 0xc4, 0xd5, 0x99, 0xc1, 0x62, 0x1c, 0x90, 0xcf, 0xa0,
-	0x4d, 0x53, 0xcd, 0x2f, 0x31, 0x51, 0x9a, 0x6a, 0x74, 0x09, 0xad, 0x59, 0x5e, 0xcb, 0x19, 0x7a,
-	0x06, 0xb7, 0x59, 0x7d, 0x09, 0xfb, 0xa9, 0x9c, 0x2a, 0x4d, 0xb3, 0x8c, 0xff, 0x8e, 0xac, 0xea,
-	0x71, 0xcf, 0x7a, 0x3c, 0xa8, 0x9a, 0xe7, 0x7e, 0x2f, 0x61, 0xb3, 0x9c, 0x7f, 0xb0, 0x6e, 0xbb,
-	0xf7, 0x30, 0x72, 0x0a, 0x89, 0x66, 0x0a, 0x89, 0xde, 0xcd, 0x18, 0xf1, 0x9c, 0x4c, 0xbe, 0x87,
-	0x26, 0xd5, 0xda, 0xbc, 0x68, 0x2e, 0x72, 0x15, 0x6c, 0x1c, 0xad, 0x1e, 0x37, 0x4e, 0x9e, 0xde,
-	0xd6, 0xfa, 0xee, 0x9c, 0x1b, 0x63, 0x2a, 0x24, 0x8b, 0x17, 0xdc, 0xc3, 0x17, 0xf0, 0xe8, 0xec,
-	0xa6, 0x0c, 0xef, 0xd4, 0x5e, 0x04, 0xc1, 0x92, 0xd3, 0x5d, 0x0a, 0x9c, 0xc0, 0xc1, 0x0d, 0x7c,
-	0x2f, 0xa2, 0x9f, 0x80, 0x2c, 0xb7, 0xd0, 0x2b, 0xea, 0xd6, 0xb2, 0x96, 0xc3, 0xb5, 0x97, 0x1a,
-	0x1d, 0xfe, 0xb3, 0x06, 0xed, 0x25, 0xa2, 0x19, 0x6f, 0x46, 0x95, 0xf6, 0xa3, 0x92, 0x98, 0xd2,
-	0x2c, 0xf5, 0x92, 0x6f, 0x19, 0x83, 0xcf, 0xce, 0xc0, 0xe4, 0x29, 0xec, 0xfc, 0x36, 0x51, 0x9a,
-	0x0f, 0xb8, 0x4d, 0x4c, 0x22, 0xbd, 0xf0, 0xca, 0x6b, 0x95, 0x78, 0xcf, 0xc2, 0x24, 0x82, 0x5d,
-	0x1b, 0xb6, 0xc2, 0xcf, 0x84, 0xb6, 0x22, 0xac, 0xc7, 0xf6, 0x8b, 0xdf, 0x96, 0x1e, 0x99, 0xd0,
-	0x25, 0x7f, 0xc0, 0x73, 0xea, 0x0b, 0x37, 0xfc, 0xfa, 0x9c, 0x7f, 0x3e, 0xb3, 0x58, 0xfe, 0x13,
-	0x68, 0xa5, 0x13, 0x69, 0x4f, 0x08, 0x9b, 0xe6, 0x54, 0xe9, 0xa9, 0xd5, 0x64, 0x3d, 0xde, 0xf6,
-	0xf0, 0x2b, 0x87, 0x92, 0x97, 0x10, 0xa4, 0x52, 0x28, 0x95, 0xf1, 0xfc, 0x82, 0xe7, 0x43, 0x53,
-	0xa7, 0xd4, 0x89, 0x1a, 0x51, 0xc9, 0xac, 0x26, 0xeb, 0xf1, 0x5e, 0xd5, 0xde, 0x33, 0xe6, 0x9e,
-	0xb1, 0x92, 0x4f, 0x61, 0x5b, 0x0b, 0x4d, 0xb3, 0x84, 0x61, 0x21, 0x14, 0xd7, 0xca, 0x2a, 0xb3,
-	0x1e, 0x6f, 0x59, 0xf4, 0x95, 0x07, 0xc9, 0x27, 0xd0, 0xf4, 0x19, 0x24, 0x0a, 0x91, 0x05, 0x1b,
-	0xee, 0x08, 0x79, 0xac, 0x87, 0xc8, 0xc8, 0x17, 0xb0, 0x5f, 0xa5, 0x24, 0xb6, 0x52, 0x89, 0x0a,
-	0x75, 0xb0, 0x69, 0x43, 0xde, 0xaf, 0xb0, 0xdf, 0x50, 0xa5, 0x63, 0x63, 0x23, 0xef, 0xa0, 0x5d,
-	0xa6, 0x66, 0x26, 0x23, 0x24, 0x53, 0x01, 0x58, 0x81, 0x3f, 0xb9, 0x5d, 0x09, 0xde, 0xc1, 0xcb,
-	0x7b, 0x27, 0x5d, 0x04, 0x14, 0xf9, 0x1a, 0xe0, 0x92, 0x66, 0x9c, 0x51, 0x2d, 0xa4, 0x0a, 0x1a,
-	0x77, 0x87, 0xfb, 0x71, 0xc6, 0xf4, 0xe1, 0x2a, 0xae, 0xe4, 0x67, 0x68, 0xf3, 0x9c, 0xf1, 0x14,
-	0x55, 0x32, 0x10, 0xd2, 0xce, 0x4b, 0x05, 0x4d, 0x1b, 0x2f, 0xba, 0x2d, 0x9e, 0xed, 0x6c, 0x37,
-	0x67, 0x67, 0x62, 0x3c, 0xe6, 0x5a, 0x23, 0x76, 0xa5, 0xa4, 0xd3, 0xb8, 0xe5, 0x03, 0x9d, 0x0b,
-	0x69, 0xa6, 0xab, 0xc2, 0x3f, 0x60, 0xef, 0x66, 0x2a, 0x61, 0x70, 0x40, 0xcd, 0x83, 0x1b, 0x61,
-	0x42, 0x73, 0x96, 0xa4, 0x33, 0x46, 0x50, 0xbb, 0xfb, 0xf4, 0x2f, 0x85, 0x8c, 0xf7, 0x6c, 0xac,
-	0x25, 0xdc, 0xdc, 0x3e, 0xdd, 0xc5, 0xdd, 0xf6, 0xa1, 0xdb, 0xa7, 0x42, 0xbf, 0xeb, 0xec, 0xbf,
-	0x81, 0xf6, 0xd2, 0xd7, 0xc8, 0x01, 0x6c, 0xb8, 0x6a, 0x38, 0xf3, 0x47, 0x6f, 0xdd, 0xbe, 0xbf,
-	0x66, 0xe4, 0x23, 0xd8, 0x9c, 0x97, 0xb7, 0x72, 0xb4, 0x7a, 0xbc, 0x15, 0xcf, 0x81, 0xf0, 0x57,
-	0xd8, 0x5d, 0xf8, 0xae, 0xdf, 0x21, 0xe7, 0xd0, 0xac, 0xae, 0xec, 0x0f, 0xdd, 0x47, 0xd5, 0x10,
-	0x8d, 0xca, 0x4a, 0x0f, 0xff, 0xae, 0x41, 0xa3, 0x62, 0x24, 0xbf, 0xc0, 0xfd, 0x02, 0x73, 0x66,
-	0x8e, 0xd1, 0xc2, 0xd2, 0xad, 0xfd, 0xdf, 0xa5, 0xbb, 0xeb, 0xc3, 0x54, 0x2c, 0xca, 0xac, 0x00,
-	0x89, 0xa9, 0x39, 0xd1, 0xf6, 0x4e, 0xb4, 0xd7, 0x06, 0x2a, 0x5b, 0x74, 0x33, 0x6e, 0x3b, 0x53,
-	0xf9, 0x33, 0x80, 0x2a, 0xfc, 0x6b, 0x05, 0x5a, 0xd7, 0xf4, 0x49, 0x1e, 0x01, 0x14, 0x93, 0x7e,
-	0xc6, 0xd3, 0xe4, 0x02, 0xa7, 0xbe, 0x97, 0x9b, 0x0e, 0xf9, 0x0e, 0xa7, 0x66, 0x81, 0x5d, 0x71,
-	0x3d, 0x62, 0x92, 0x5e, 0xd1, 0xcc, 0x2f, 0x01, 0xbf, 0xc0, 0xe6, 0xb8, 0x3b, 0xfd, 0xcf, 0x80,
-	0x54, 0xa8, 0x94, 0x31, 0x89, 0x4a, 0xf9, 0x4b, 0xb4, 0x3d, 0xb7, 0x74, 0x9d, 0x81, 0x7c, 0x0e,
-	0x6d, 0x7f, 0xdd, 0xba, 0xe9, 0x8c, 0x31, 0xd7, 0xfe, 0x36, 0xdd, 0x71, 0x86, 0xb3, 0x12, 0x27,
-	0x01, 0xac, 0xf7, 0x69, 0x46, 0xf3, 0x14, 0xfd, 0xd2, 0x9a, 0xbd, 0x9a, 0x5b, 0xdb, 0x2d, 0xa8,
-	0xd9, 0x52, 0x73, 0x2b, 0xaa, 0x69, 0xc1, 0xd9, 0x4a, 0x3b, 0x84, 0x06, 0xe6, 0xac, 0xa4, 0xb8,
-	0xad, 0x04, 0x98, 0x33, 0x4f, 0x08, 0xff, 0xad, 0x41, 0x7b, 0xa9, 0xe9, 0x46, 0x8e, 0x76, 0xa7,
-	0xba, 0xae, 0xd8, 0xe7, 0x05, 0xe5, 0xad, 0x2c, 0x2a, 0xef, 0x18, 0x76, 0x9c, 0x69, 0x3e, 0x0d,
-	0x5f, 0xfe, 0xb6, 0xc5, 0xcb, 0x51, 0x98, 0xda, 0x9d, 0x1c, 0x50, 0x26, 0x7d, 0xae, 0x07, 0x1c,
-	0x33, 0x36, 0xab, 0x7d, 0x66, 0x38, 0xf5, 0x38, 0x39, 0x81, 0x07, 0xa2, 0x9f, 0xf1, 0xf7, 0x13,
-	0x4c, 0x2a, 0x7f, 0x38, 0xa8, 0x82, 0x35, 0x3b, 0xe7, 0x5d, 0x6f, 0x7c, 0x5b, 0xfe, 0xea, 0xa0,
-	0x32, 0x5d, 0xa1, 0xc3, 0xa1, 0xc4, 0xa1, 0xb9, 0xa2, 0x14, 0x1f, 0x06, 0xf7, 0x8e, 0x56, 0x4d,
-	0x57, 0x4a, 0xb0, 0xc7, 0x87, 0xe1, 0x6b, 0x68, 0x5d, 0x5b, 0x7e, 0xa6, 0xcf, 0xb3, 0x26, 0xf9,
-	0x63, 0xe5, 0x5f, 0xcd, 0xb1, 0xb2, 0x65, 0xd9, 0xaa, 0x56, 0x6c, 0xaa, 0x73, 0xa0, 0x7f, 0xcf,
-	0xfe, 0x73, 0xbc, 0xf8, 0x2f, 0x00, 0x00, 0xff, 0xff, 0x4b, 0x1d, 0x03, 0x9c, 0xe1, 0x0a, 0x00,
-	0x00,
+var fileDescriptor_messages_bd1ed2903542c11b = []byte{
+	// 1287 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x56, 0xdd, 0x72, 0xdb, 0x44,
+	0x14, 0xc6, 0x8e, 0xf3, 0x77, 0xec, 0xd4, 0xf6, 0xa6, 0x4d, 0x95, 0x42, 0x9b, 0xa0, 0xc2, 0x34,
+	0x2d, 0x53, 0x67, 0x9a, 0x0e, 0x4c, 0xb9, 0x94, 0x5d, 0x97, 0x86, 0xa6, 0x4e, 0x91, 0x9c, 0x42,
+	0x3b, 0x30, 0x9a, 0xb5, 0xb4, 0xb6, 0x45, 0x64, 0xad, 0xba, 0xbb, 0x4e, 0xc6, 0x5c, 0x70, 0xc7,
+	0x33, 0x70, 0xc1, 0x13, 0xf0, 0x32, 0x3c, 0x0c, 0x4f, 0xc0, 0xec, 0x6a, 0x25, 0xcb, 0x3f, 0x49,
+	0x87, 0x3b, 0xe9, 0xfb, 0xbe, 0x73, 0x74, 0xfe, 0xf6, 0xac, 0xc0, 0x8c, 0x19, 0x15, 0xf4, 0xb0,
+	0x47, 0xb0, 0x47, 0xa3, 0xc3, 0xf8, 0x28, 0x3e, 0xbc, 0x78, 0x72, 0x38, 0x22, 0x9c, 0xe3, 0x01,
+	0xe1, 0x0d, 0x45, 0xa2, 0x1d, 0x22, 0x86, 0x84, 0x91, 0xf1, 0xa8, 0x91, 0xc8, 0x1a, 0xf1, 0x51,
+	0xdc, 0xb8, 0x78, 0x72, 0x67, 0x6f, 0x40, 0xe9, 0x20, 0x24, 0x87, 0x4a, 0xd5, 0x1b, 0xf7, 0x0f,
+	0x45, 0x30, 0x22, 0x5c, 0xe0, 0x51, 0x9c, 0x18, 0x9a, 0x8f, 0xe1, 0x76, 0x53, 0x59, 0x34, 0x43,
+	0xea, 0x9d, 0xbf, 0xc4, 0x7c, 0x68, 0x45, 0x11, 0x1d, 0x47, 0x1e, 0x41, 0x08, 0x4a, 0x43, 0xcc,
+	0x87, 0x46, 0x61, 0xbf, 0x70, 0x50, 0xb1, 0xd5, 0xb3, 0x79, 0x00, 0x28, 0x27, 0xb7, 0xc9, 0x87,
+	0x31, 0xe1, 0x62, 0xa9, 0xd2, 0x82, 0x7b, 0x8b, 0xca, 0xe6, 0xc4, 0x09, 0xa9, 0xe8, 0x8c, 0x47,
+	0x3d, 0xc2, 0xd0, 0x1e, 0x94, 0x79, 0x48, 0x85, 0x1b, 0xa9, 0x57, 0x65, 0x5c, 0xb2, 0x81, 0x67,
+	0x02, 0xf3, 0x0d, 0x6c, 0xcf, 0xb8, 0xe0, 0x31, 0x8d, 0x38, 0x41, 0xdf, 0xc2, 0x6a, 0x4f, 0x02,
+	0xca, 0xa2, 0x7c, 0x74, 0xbf, 0xb1, 0x3c, 0xf7, 0x46, 0xde, 0x36, 0xb1, 0x30, 0xff, 0x58, 0x81,
+	0x72, 0x0e, 0x96, 0x21, 0xc4, 0x98, 0x91, 0x48, 0xb8, 0xb9, 0xf8, 0x21, 0x81, 0x64, 0x2d, 0xe6,
+	0x63, 0x2c, 0xce, 0xc7, 0x88, 0xee, 0xc3, 0x16, 0xc3, 0x91, 0x8f, 0xa9, 0xcb, 0xc8, 0x05, 0xc1,
+	0xa1, 0xb1, 0xa2, 0x7c, 0x54, 0x12, 0xd0, 0x56, 0x18, 0x32, 0x61, 0x2b, 0xa6, 0x97, 0xae, 0x37,
+	0xc4, 0x41, 0xe4, 0x32, 0xd2, 0x37, 0x4a, 0x4a, 0x54, 0x8e, 0xe9, 0x65, 0x4b, 0x62, 0x36, 0xe9,
+	0xa3, 0x47, 0x50, 0xc7, 0x9e, 0x08, 0x2e, 0x88, 0xcb, 0x05, 0x16, 0x24, 0x09, 0x68, 0x55, 0xe9,
+	0xaa, 0x09, 0xe1, 0x48, 0x5c, 0x45, 0xf5, 0x0d, 0xdc, 0xf6, 0xd8, 0x84, 0x0b, 0x1c, 0x86, 0xc1,
+	0x6f, 0xc4, 0xcf, 0x5b, 0xac, 0x29, 0x8b, 0x5b, 0x79, 0x7a, 0x6a, 0xf7, 0x0c, 0x36, 0xb3, 0xfe,
+	0x1b, 0xeb, 0xaa, 0x7a, 0x77, 0x1a, 0xc9, 0x84, 0x34, 0xd2, 0x09, 0x69, 0x74, 0x53, 0x85, 0x3d,
+	0x15, 0xa3, 0xd7, 0x50, 0xc1, 0x42, 0xc8, 0x17, 0x11, 0xd0, 0x88, 0x1b, 0x1b, 0xfb, 0x2b, 0x07,
+	0xe5, 0xa3, 0x87, 0x57, 0x95, 0xde, 0x9a, 0x6a, 0x6d, 0xe2, 0x51, 0xe6, 0xdb, 0x33, 0xe6, 0xe6,
+	0x53, 0xb8, 0xdb, 0x5a, 0x16, 0xe1, 0xb5, 0xb3, 0xd7, 0x00, 0x63, 0xc1, 0xe8, 0xba, 0x09, 0x1c,
+	0xc3, 0xee, 0x12, 0xbd, 0x1e, 0xa2, 0x9f, 0x00, 0x2d, 0x96, 0x50, 0x4f, 0xd4, 0x95, 0x69, 0x2d,
+	0xba, 0xab, 0x2f, 0x14, 0xda, 0xfc, 0x7b, 0x15, 0xea, 0x0b, 0x42, 0xd9, 0xde, 0x10, 0x73, 0xa1,
+	0x5b, 0xc5, 0x88, 0x87, 0x43, 0x4f, 0x8f, 0x7c, 0x55, 0x12, 0x3a, 0x3a, 0x09, 0xa3, 0x87, 0x50,
+	0xfb, 0x75, 0xcc, 0x45, 0xd0, 0x0f, 0x54, 0x60, 0x8c, 0xe0, 0x73, 0x3d, 0x79, 0xd5, 0x0c, 0x77,
+	0x14, 0x8c, 0x1a, 0xb0, 0xad, 0xdc, 0xe6, 0xf4, 0x21, 0x15, 0x6a, 0x08, 0x4b, 0xb6, 0xfa, 0xe2,
+	0xf7, 0x99, 0x45, 0x48, 0x45, 0xa6, 0xef, 0x07, 0x11, 0xd6, 0x89, 0x4b, 0x7d, 0x69, 0xaa, 0x7f,
+	0x91, 0x32, 0x4a, 0xff, 0x00, 0xaa, 0xde, 0x98, 0xa9, 0x13, 0xe2, 0x4f, 0x22, 0xcc, 0xc5, 0x44,
+	0xcd, 0x64, 0xc9, 0xbe, 0xa1, 0xe1, 0xe7, 0x09, 0x8a, 0x9e, 0x81, 0xe1, 0x31, 0xca, 0x79, 0x18,
+	0x44, 0xe7, 0x41, 0x34, 0x90, 0x79, 0x32, 0xe1, 0xf2, 0x21, 0x66, 0xbe, 0x9a, 0xc9, 0x92, 0xbd,
+	0x93, 0xe7, 0x1d, 0x49, 0x3b, 0x92, 0x45, 0x5f, 0xc2, 0x0d, 0x41, 0x05, 0x0e, 0x5d, 0x9f, 0xc4,
+	0x94, 0x07, 0x82, 0xab, 0xc9, 0x2c, 0xd9, 0x5b, 0x0a, 0x7d, 0xae, 0x41, 0xf4, 0x39, 0x54, 0x74,
+	0x04, 0x2e, 0x27, 0xc4, 0x37, 0x36, 0x92, 0x23, 0xa4, 0x31, 0x87, 0x10, 0x1f, 0x7d, 0x0d, 0xb7,
+	0xf3, 0x12, 0x57, 0x65, 0xca, 0x08, 0x27, 0xc2, 0xd8, 0x54, 0x2e, 0x6f, 0xe6, 0xd4, 0x27, 0x98,
+	0x0b, 0x5b, 0x72, 0xa8, 0x0b, 0xf5, 0x2c, 0x34, 0xd9, 0x19, 0xca, 0x7c, 0x6e, 0x80, 0x1a, 0xf0,
+	0x07, 0x57, 0x4f, 0x82, 0x36, 0xd0, 0xe3, 0x5d, 0xf3, 0x66, 0x01, 0x8e, 0xbe, 0x03, 0xb8, 0xc0,
+	0x61, 0xe0, 0x63, 0x41, 0x19, 0x37, 0xca, 0xd7, 0xbb, 0x7b, 0x9b, 0x2a, 0xb5, 0xbb, 0x9c, 0x29,
+	0x7a, 0x0f, 0xf5, 0x20, 0xf2, 0x03, 0x8f, 0x70, 0xb7, 0x4f, 0x99, 0xea, 0x17, 0x37, 0x2a, 0xca,
+	0x5f, 0xe3, 0x2a, 0x7f, 0xaa, 0xb2, 0x56, 0xe4, 0xb7, 0xe8, 0x68, 0x14, 0x08, 0x41, 0x88, 0xc5,
+	0x18, 0x9e, 0xd8, 0x55, 0xed, 0xe8, 0x05, 0x65, 0xb2, 0xbb, 0xdc, 0xfc, 0x1d, 0x76, 0x96, 0x4b,
+	0x91, 0x0f, 0xbb, 0x58, 0x3e, 0x24, 0x2d, 0x74, 0x71, 0xe4, 0xbb, 0x5e, 0xaa, 0x30, 0x0a, 0xd7,
+	0x9f, 0xfe, 0x05, 0x97, 0xf6, 0x8e, 0xf2, 0xb5, 0x80, 0xcb, 0xdb, 0xc7, 0x9a, 0xdd, 0x6d, 0x1f,
+	0xbb, 0x7d, 0x72, 0xf2, 0xeb, 0xce, 0xfe, 0x09, 0xd4, 0x17, 0xbe, 0x86, 0x76, 0x61, 0x23, 0xc9,
+	0x26, 0xf0, 0xf5, 0xd1, 0x5b, 0x57, 0xef, 0xc7, 0x3e, 0xfa, 0x0c, 0x36, 0xa7, 0xe9, 0x15, 0xf7,
+	0x57, 0x0e, 0xb6, 0xec, 0x29, 0x60, 0xfe, 0x02, 0xdb, 0x33, 0xdf, 0xd5, 0x3b, 0xe4, 0x05, 0x54,
+	0xf2, 0x2b, 0xfb, 0x63, 0xf7, 0x51, 0xde, 0x45, 0x39, 0xb7, 0xd2, 0xcd, 0xbf, 0x0a, 0x50, 0xce,
+	0x91, 0xe8, 0x67, 0xb8, 0x19, 0x93, 0xc8, 0x97, 0xc7, 0x68, 0x66, 0xe9, 0x16, 0xfe, 0xef, 0xd2,
+	0xdd, 0xd6, 0x6e, 0x72, 0x0c, 0x97, 0x2b, 0x80, 0x11, 0x4f, 0x9e, 0x68, 0x75, 0x27, 0xaa, 0x6b,
+	0x83, 0x70, 0x95, 0x74, 0xc5, 0xae, 0x27, 0x54, 0xf6, 0x33, 0x40, 0xb8, 0xf9, 0x67, 0x11, 0xaa,
+	0x73, 0xf3, 0x89, 0xee, 0x02, 0xc4, 0xe3, 0x5e, 0x18, 0x78, 0xee, 0x39, 0x99, 0xe8, 0x5a, 0x6e,
+	0x26, 0xc8, 0x2b, 0x32, 0x91, 0x0b, 0xec, 0x32, 0x10, 0x43, 0x9f, 0xe1, 0x4b, 0x1c, 0xea, 0x25,
+	0xa0, 0x17, 0xd8, 0x14, 0x4f, 0x4e, 0xff, 0x63, 0x40, 0x39, 0x29, 0xf6, 0x7d, 0x46, 0x38, 0xd7,
+	0x97, 0x68, 0x7d, 0xca, 0x58, 0x09, 0x81, 0xbe, 0x82, 0xba, 0xbe, 0x6e, 0x93, 0xee, 0x8c, 0x48,
+	0x24, 0xf4, 0x6d, 0x5a, 0x4b, 0x88, 0x56, 0x86, 0x23, 0x03, 0xd6, 0x7b, 0x38, 0xc4, 0x91, 0x47,
+	0xf4, 0xd2, 0x4a, 0x5f, 0xe5, 0xad, 0x9d, 0x2c, 0xa8, 0x74, 0xa9, 0x25, 0x2b, 0xaa, 0xa2, 0xc0,
+	0x74, 0xa5, 0xed, 0x41, 0x99, 0x44, 0x7e, 0x26, 0x49, 0xb6, 0x12, 0x90, 0xc8, 0xd7, 0x02, 0xf3,
+	0xdf, 0x02, 0xd4, 0x17, 0x8a, 0x2e, 0xc7, 0x51, 0xed, 0xd4, 0xa4, 0x2a, 0xea, 0x79, 0x66, 0xf2,
+	0x8a, 0xb3, 0x93, 0x77, 0x00, 0xb5, 0x84, 0x9a, 0x76, 0x43, 0xa7, 0x7f, 0x43, 0xe1, 0x59, 0x2b,
+	0x64, 0xee, 0xc9, 0x38, 0x10, 0xe6, 0xf6, 0x02, 0xd1, 0x0f, 0x48, 0xe8, 0xa7, 0xb9, 0xa7, 0x44,
+	0x53, 0xe3, 0xe8, 0x08, 0x6e, 0xd1, 0x5e, 0x18, 0x7c, 0x18, 0x13, 0x37, 0xf7, 0x87, 0x43, 0xb8,
+	0xb1, 0xaa, 0xfa, 0xbc, 0xad, 0xc9, 0x37, 0xd9, 0xaf, 0x0e, 0xe1, 0xb2, 0x2a, 0x78, 0x30, 0x60,
+	0x64, 0x20, 0xaf, 0x28, 0x1e, 0x0c, 0x8c, 0xb5, 0xfd, 0x15, 0x59, 0x95, 0x0c, 0x74, 0x82, 0x81,
+	0x79, 0x0c, 0xd5, 0xb9, 0xe5, 0x27, 0xeb, 0x9c, 0x16, 0x49, 0x1f, 0x2b, 0xfd, 0x2a, 0x8f, 0x95,
+	0x4a, 0x4b, 0x65, 0x55, 0x54, 0xa1, 0x4e, 0x81, 0x47, 0xff, 0x14, 0x61, 0xb5, 0x4b, 0xe3, 0xc0,
+	0x43, 0x65, 0x58, 0x3f, 0xeb, 0xbc, 0xea, 0x9c, 0xfe, 0xd8, 0xa9, 0x7d, 0x82, 0xee, 0xc1, 0x9d,
+	0x66, 0xdb, 0x6a, 0x9d, 0x76, 0xdc, 0xe6, 0xc9, 0x69, 0xeb, 0x95, 0xfb, 0xd2, 0x72, 0x5e, 0xba,
+	0x56, 0xa7, 0x73, 0x7a, 0xd6, 0x69, 0xb5, 0x6b, 0x05, 0x64, 0xc0, 0xcd, 0x19, 0xde, 0x6e, 0xff,
+	0x70, 0xd6, 0x76, 0xba, 0xb5, 0x22, 0x7a, 0x00, 0xf7, 0x97, 0x31, 0x6e, 0xf3, 0x9d, 0xeb, 0x9c,
+	0x9c, 0x76, 0xdd, 0xce, 0xd9, 0xeb, 0x66, 0xdb, 0xae, 0xad, 0xa0, 0x5d, 0xb8, 0x35, 0x27, 0x74,
+	0xde, 0x9c, 0x76, 0x9c, 0x76, 0xad, 0x84, 0xbe, 0x80, 0xfd, 0x96, 0xfd, 0xce, 0xe9, 0x5a, 0x27,
+	0x27, 0xc7, 0xef, 0xdb, 0xcf, 0x5d, 0xa7, 0x6b, 0x75, 0xdb, 0x73, 0x31, 0xac, 0xca, 0x18, 0x97,
+	0xa8, 0xd2, 0x48, 0xd6, 0xd0, 0x1e, 0x7c, 0xba, 0x94, 0xd7, 0x9f, 0x59, 0x97, 0x0e, 0xac, 0x56,
+	0xf7, 0xf8, 0x6d, 0x7b, 0xe9, 0x07, 0x36, 0x64, 0x92, 0x33, 0x7c, 0xea, 0x7a, 0x53, 0xc6, 0x3e,
+	0xc7, 0x68, 0xa7, 0xd0, 0x5b, 0x53, 0x3f, 0x71, 0x4f, 0xff, 0x0b, 0x00, 0x00, 0xff, 0xff, 0xf2,
+	0x6f, 0xfa, 0x29, 0x32, 0x0c, 0x00, 0x00,
 }
--- a/proto/beacon/p2p/v1/messages.proto
+++ b/proto/beacon/p2p/v1/messages.proto
@@ -4,6 +4,20 @@ package ethereum.beacon.p2p.v1;

 import "google/protobuf/timestamp.proto";

+enum Topic {
+  UNKNOWN = 0;
+  BEACON_BLOCK_HASH_ANNOUNCE = 1;
+  BEACON_BLOCK_REQUEST = 2;
+  BEACON_BLOCK_REQUEST_BY_SLOT_NUMBER = 3;
+  BEACON_BLOCK_RESPONSE = 4;
+  CRYSTALLIZED_STATE_HASH_ANNOUNCE = 5;
+  CRYSTALLIZED_STATE_REQUEST = 6;
+  CRYSTALLIZED_STATE_RESPONSE = 7;
+  ACTIVE_STATE_HASH_ANNOUNCE = 8;
+  ACTIVE_STATE_REQUEST = 9;
+  ACTIVE_STATE_RESPONSE = 10;
+} 
+
 message BeaconBlockHashAnnounce {
  bytes hash = 1;
 }
--- a/proto/sharding/p2p/v1/messages.pb.go
+++ b/proto/sharding/p2p/v1/messages.pb.go
@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
-// source: messages.proto
+// source: proto/sharding/p2p/v1/messages.proto

-package ethereum_sharding_p2p_v1
+package v1

 import proto "github.com/golang/protobuf/proto"
 import fmt "fmt"
@@ -18,69 +18,38 @@ var _ = math.Inf
 // proto package needs to be updated.
 const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package

-// TODO: Split the topics into p2p for beacon chain and p2p for sharding.
 type Topic int32

 const (
-	Topic_UNKNOWN                             Topic = 0
-	Topic_COLLATION_BODY_REQUEST              Topic = 1
-	Topic_COLLATION_BODY_RESPONSE             Topic = 2
-	Topic_TRANSACTIONS                        Topic = 3
-	Topic_BEACON_BLOCK_HASH_ANNOUNCE          Topic = 4
-	Topic_BEACON_BLOCK_REQUEST                Topic = 5
-	Topic_BEACON_BLOCK_REQUEST_BY_SLOT_NUMBER Topic = 6
-	Topic_BEACON_BLOCK_RESPONSE               Topic = 7
-	Topic_CRYSTALLIZED_STATE_HASH_ANNOUNCE    Topic = 8
-	Topic_CRYSTALLIZED_STATE_REQUEST          Topic = 9
-	Topic_CRYSTALLIZED_STATE_RESPONSE         Topic = 10
-	Topic_ACTIVE_STATE_HASH_ANNOUNCE          Topic = 11
-	Topic_ACTIVE_STATE_REQUEST                Topic = 12
-	Topic_ACTIVE_STATE_RESPONSE               Topic = 13
+	Topic_UNKNOWN                 Topic = 0
+	Topic_COLLATION_BODY_REQUEST  Topic = 1
+	Topic_COLLATION_BODY_RESPONSE Topic = 2
+	Topic_TRANSACTIONS            Topic = 3
 )

 var Topic_name = map[int32]string{
-	0:  "UNKNOWN",
-	1:  "COLLATION_BODY_REQUEST",
-	2:  "COLLATION_BODY_RESPONSE",
-	3:  "TRANSACTIONS",
-	4:  "BEACON_BLOCK_HASH_ANNOUNCE",
-	5:  "BEACON_BLOCK_REQUEST",
-	6:  "BEACON_BLOCK_REQUEST_BY_SLOT_NUMBER",
-	7:  "BEACON_BLOCK_RESPONSE",
-	8:  "CRYSTALLIZED_STATE_HASH_ANNOUNCE",
-	9:  "CRYSTALLIZED_STATE_REQUEST",
-	10: "CRYSTALLIZED_STATE_RESPONSE",
-	11: "ACTIVE_STATE_HASH_ANNOUNCE",
-	12: "ACTIVE_STATE_REQUEST",
-	13: "ACTIVE_STATE_RESPONSE",
+	0: "UNKNOWN",
+	1: "COLLATION_BODY_REQUEST",
+	2: "COLLATION_BODY_RESPONSE",
+	3: "TRANSACTIONS",
 }
 var Topic_value = map[string]int32{
-	"UNKNOWN":                             0,
-	"COLLATION_BODY_REQUEST":              1,
-	"COLLATION_BODY_RESPONSE":             2,
-	"TRANSACTIONS":                        3,
-	"BEACON_BLOCK_HASH_ANNOUNCE":          4,
-	"BEACON_BLOCK_REQUEST":                5,
-	"BEACON_BLOCK_REQUEST_BY_SLOT_NUMBER": 6,
-	"BEACON_BLOCK_RESPONSE":               7,
-	"CRYSTALLIZED_STATE_HASH_ANNOUNCE":    8,
-	"CRYSTALLIZED_STATE_REQUEST":          9,
-	"CRYSTALLIZED_STATE_RESPONSE":         10,
-	"ACTIVE_STATE_HASH_ANNOUNCE":          11,
-	"ACTIVE_STATE_REQUEST":                12,
-	"ACTIVE_STATE_RESPONSE":               13,
+	"UNKNOWN":                 0,
+	"COLLATION_BODY_REQUEST":  1,
+	"COLLATION_BODY_RESPONSE": 2,
+	"TRANSACTIONS":            3,
 }

 func (x Topic) String() string {
 	return proto.EnumName(Topic_name, int32(x))
 }
 func (Topic) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_messages_2d61a34a4bdb49bc, []int{0}
+	return fileDescriptor_messages_8b1affe159ebec96, []int{0}
 }

 type CollationBodyRequest struct {
-	ShardId              uint64   `protobuf:"varint,1,opt,name=shard_id,json=shardId,proto3" json:"shard_id,omitempty"`
-	Period               uint64   `protobuf:"varint,2,opt,name=period,proto3" json:"period,omitempty"`
+	ShardId              uint64   `protobuf:"varint,1,opt,name=shard_id,json=shardId" json:"shard_id,omitempty"`
+	Period               uint64   `protobuf:"varint,2,opt,name=period" json:"period,omitempty"`
 	ChunkRoot            []byte   `protobuf:"bytes,3,opt,name=chunk_root,json=chunkRoot,proto3" json:"chunk_root,omitempty"`
 	ProposerAddress      []byte   `protobuf:"bytes,4,opt,name=proposer_address,json=proposerAddress,proto3" json:"proposer_address,omitempty"`
 	Signature            []byte   `protobuf:"bytes,5,opt,name=signature,proto3" json:"signature,omitempty"`
@@ -93,7 +62,7 @@ func (m *CollationBodyRequest) Reset()         { *m = CollationBodyRequest{} }
 func (m *CollationBodyRequest) String() string { return proto.CompactTextString(m) }
 func (*CollationBodyRequest) ProtoMessage()    {}
 func (*CollationBodyRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_2d61a34a4bdb49bc, []int{0}
+	return fileDescriptor_messages_8b1affe159ebec96, []int{0}
 }
 func (m *CollationBodyRequest) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_CollationBodyRequest.Unmarshal(m, b)
@@ -160,7 +129,7 @@ func (m *CollationBodyResponse) Reset()         { *m = CollationBodyResponse{} }
 func (m *CollationBodyResponse) String() string { return proto.CompactTextString(m) }
 func (*CollationBodyResponse) ProtoMessage()    {}
 func (*CollationBodyResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_2d61a34a4bdb49bc, []int{1}
+	return fileDescriptor_messages_8b1affe159ebec96, []int{1}
 }
 func (m *CollationBodyResponse) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_CollationBodyResponse.Unmarshal(m, b)
@@ -195,13 +164,13 @@ func (m *CollationBodyResponse) GetBody() []byte {
 }

 type Transaction struct {
-	Nonce                uint64     `protobuf:"varint,1,opt,name=nonce,proto3" json:"nonce,omitempty"`
-	GasPrice             uint64     `protobuf:"varint,2,opt,name=gas_price,json=gasPrice,proto3" json:"gas_price,omitempty"`
-	GasLimit             uint64     `protobuf:"varint,3,opt,name=gas_limit,json=gasLimit,proto3" json:"gas_limit,omitempty"`
+	Nonce                uint64     `protobuf:"varint,1,opt,name=nonce" json:"nonce,omitempty"`
+	GasPrice             uint64     `protobuf:"varint,2,opt,name=gas_price,json=gasPrice" json:"gas_price,omitempty"`
+	GasLimit             uint64     `protobuf:"varint,3,opt,name=gas_limit,json=gasLimit" json:"gas_limit,omitempty"`
 	Recipient            []byte     `protobuf:"bytes,4,opt,name=recipient,proto3" json:"recipient,omitempty"`
-	Value                uint64     `protobuf:"varint,5,opt,name=value,proto3" json:"value,omitempty"`
+	Value                uint64     `protobuf:"varint,5,opt,name=value" json:"value,omitempty"`
 	Input                []byte     `protobuf:"bytes,6,opt,name=input,proto3" json:"input,omitempty"`
-	Signature            *Signature `protobuf:"bytes,7,opt,name=signature,proto3" json:"signature,omitempty"`
+	Signature            *Signature `protobuf:"bytes,7,opt,name=signature" json:"signature,omitempty"`
 	XXX_NoUnkeyedLiteral struct{}   `json:"-"`
 	XXX_unrecognized     []byte     `json:"-"`
 	XXX_sizecache        int32      `json:"-"`
@@ -211,7 +180,7 @@ func (m *Transaction) Reset()         { *m = Transaction{} }
 func (m *Transaction) String() string { return proto.CompactTextString(m) }
 func (*Transaction) ProtoMessage()    {}
 func (*Transaction) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_2d61a34a4bdb49bc, []int{2}
+	return fileDescriptor_messages_8b1affe159ebec96, []int{2}
 }
 func (m *Transaction) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Transaction.Unmarshal(m, b)
@@ -281,9 +250,9 @@ func (m *Transaction) GetSignature() *Signature {
 }

 type Signature struct {
-	V                    uint64   `protobuf:"varint,1,opt,name=v,proto3" json:"v,omitempty"`
-	R                    uint64   `protobuf:"varint,2,opt,name=r,proto3" json:"r,omitempty"`
-	S                    uint64   `protobuf:"varint,3,opt,name=s,proto3" json:"s,omitempty"`
+	V                    uint64   `protobuf:"varint,1,opt,name=v" json:"v,omitempty"`
+	R                    uint64   `protobuf:"varint,2,opt,name=r" json:"r,omitempty"`
+	S                    uint64   `protobuf:"varint,3,opt,name=s" json:"s,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -293,7 +262,7 @@ func (m *Signature) Reset()         { *m = Signature{} }
 func (m *Signature) String() string { return proto.CompactTextString(m) }
 func (*Signature) ProtoMessage()    {}
 func (*Signature) Descriptor() ([]byte, []int) {
-	return fileDescriptor_messages_2d61a34a4bdb49bc, []int{3}
+	return fileDescriptor_messages_8b1affe159ebec96, []int{3}
 }
 func (m *Signature) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Signature.Unmarshal(m, b)
@@ -342,44 +311,39 @@ func init() {
 	proto.RegisterEnum("ethereum.sharding.p2p.v1.Topic", Topic_name, Topic_value)
 }

-func init() { proto.RegisterFile("messages.proto", fileDescriptor_messages_2d61a34a4bdb49bc) }
-
-var fileDescriptor_messages_2d61a34a4bdb49bc = []byte{
-	// 569 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x74, 0x53, 0xed, 0x4e, 0xd4, 0x4c,
-	0x18, 0x7d, 0x0b, 0xfb, 0xc1, 0x3e, 0xdb, 0x57, 0x9b, 0x09, 0x60, 0x01, 0x95, 0x0d, 0x98, 0x88,
-	0xfe, 0x68, 0x22, 0xc6, 0x0b, 0xe8, 0x96, 0x26, 0x10, 0x6a, 0x8b, 0x6d, 0x57, 0x83, 0x7f, 0x26,
-	0x43, 0x3b, 0xd9, 0x4e, 0x5c, 0x3a, 0x75, 0xa6, 0xdd, 0x84, 0x4b, 0xf1, 0x32, 0xbc, 0x2d, 0xaf,
-	0xc2, 0x4c, 0x3f, 0x10, 0xd6, 0xf5, 0xdf, 0x9e, 0x8f, 0x9c, 0xe7, 0x9c, 0xd9, 0x14, 0x9e, 0xdc,
-	0x52, 0x29, 0xc9, 0x9c, 0x4a, 0xab, 0x10, 0xbc, 0xe4, 0xc8, 0xa4, 0x65, 0x46, 0x05, 0xad, 0x6e,
-	0x2d, 0x99, 0x11, 0x91, 0xb2, 0x7c, 0x6e, 0x15, 0xa7, 0x85, 0xb5, 0x7c, 0x77, 0xf4, 0x53, 0x83,
-	0x6d, 0x87, 0x2f, 0x16, 0xa4, 0x64, 0x3c, 0x9f, 0xf2, 0xf4, 0x2e, 0xa4, 0xdf, 0x2b, 0x2a, 0x4b,
-	0xb4, 0x07, 0x5b, 0xb5, 0x17, 0xb3, 0xd4, 0xd4, 0x26, 0xda, 0x49, 0x2f, 0x1c, 0xd6, 0xf8, 0x22,
-	0x45, 0xbb, 0x30, 0x28, 0xa8, 0x60, 0x3c, 0x35, 0x37, 0x6a, 0xa1, 0x45, 0xe8, 0x05, 0x40, 0x92,
-	0x55, 0xf9, 0x37, 0x2c, 0x38, 0x2f, 0xcd, 0xcd, 0x89, 0x76, 0xa2, 0x87, 0xa3, 0x9a, 0x09, 0x39,
-	0x2f, 0xd1, 0x1b, 0x30, 0x0a, 0xc1, 0x0b, 0x2e, 0xa9, 0xc0, 0x24, 0x4d, 0x05, 0x95, 0xd2, 0xec,
-	0xd5, 0xa6, 0xa7, 0x1d, 0x6f, 0x37, 0x34, 0x7a, 0x0e, 0x23, 0xc9, 0xe6, 0x39, 0x29, 0x2b, 0x41,
-	0xcd, 0x7e, 0x13, 0x74, 0x4f, 0x1c, 0x79, 0xb0, 0xb3, 0x52, 0x59, 0x16, 0x3c, 0x97, 0x14, 0x1d,
-	0xc2, 0x38, 0xa3, 0x24, 0xa5, 0x02, 0x67, 0x44, 0x66, 0x75, 0x6d, 0x3d, 0x84, 0x86, 0x3a, 0x27,
-	0x32, 0x43, 0x08, 0x7a, 0x37, 0x3c, 0xbd, 0xab, 0x7b, 0xeb, 0x61, 0xfd, 0xfb, 0xe8, 0x97, 0x06,
-	0xe3, 0x58, 0x90, 0x5c, 0x92, 0x44, 0x05, 0xa2, 0x6d, 0xe8, 0xe7, 0x3c, 0x4f, 0x68, 0xbb, 0xba,
-	0x01, 0xe8, 0x00, 0x46, 0x73, 0x22, 0x71, 0x21, 0x58, 0x42, 0xdb, 0xd9, 0x5b, 0x73, 0x22, 0xaf,
-	0x14, 0xee, 0xc4, 0x05, 0xbb, 0x65, 0xcd, 0xee, 0x46, 0xf4, 0x14, 0x56, 0x5b, 0x04, 0x4d, 0x58,
-	0xc1, 0x68, 0x5e, 0xb6, 0x7b, 0xff, 0x10, 0xea, 0xda, 0x92, 0x2c, 0xaa, 0x66, 0x65, 0x2f, 0x6c,
-	0x80, 0x62, 0x59, 0x5e, 0x54, 0xa5, 0x39, 0xa8, 0xfd, 0x0d, 0x40, 0xf6, 0xc3, 0x57, 0x19, 0x4e,
-	0xb4, 0x93, 0xf1, 0xe9, 0xb1, 0xf5, 0xaf, 0x7f, 0xd6, 0x8a, 0x3a, 0xeb, 0xc3, 0xa7, 0xfb, 0x00,
-	0xa3, 0x7b, 0x1e, 0xe9, 0xa0, 0x2d, 0xdb, 0x95, 0xda, 0x52, 0x21, 0xd1, 0x2e, 0xd3, 0x84, 0x42,
-	0xb2, 0x9d, 0xa2, 0xc9, 0xb7, 0x3f, 0x36, 0xa1, 0x1f, 0xf3, 0x82, 0x25, 0x68, 0x0c, 0xc3, 0x99,
-	0x7f, 0xe9, 0x07, 0x5f, 0x7c, 0xe3, 0x3f, 0xb4, 0x0f, 0xbb, 0x4e, 0xe0, 0x79, 0x76, 0x7c, 0x11,
-	0xf8, 0x78, 0x1a, 0x9c, 0x5d, 0xe3, 0xd0, 0xfd, 0x34, 0x73, 0xa3, 0xd8, 0xd0, 0xd0, 0x01, 0x3c,
-	0xfb, 0x4b, 0x8b, 0xae, 0x02, 0x3f, 0x72, 0x8d, 0x0d, 0x64, 0x80, 0x1e, 0x87, 0xb6, 0x1f, 0xd9,
-	0x8e, 0x92, 0x23, 0x63, 0x13, 0xbd, 0x84, 0xfd, 0xa9, 0x6b, 0x3b, 0xca, 0xeb, 0x05, 0xce, 0x25,
-	0x3e, 0xb7, 0xa3, 0x73, 0x6c, 0xfb, 0x7e, 0x30, 0xf3, 0x1d, 0xd7, 0xe8, 0x21, 0x13, 0xb6, 0x1f,
-	0xe9, 0xdd, 0xa1, 0x3e, 0x7a, 0x0d, 0xc7, 0xeb, 0x14, 0x3c, 0xbd, 0xc6, 0x91, 0x17, 0xc4, 0xd8,
-	0x9f, 0x7d, 0x9c, 0xba, 0xa1, 0x31, 0x40, 0x7b, 0xb0, 0xb3, 0x62, 0x6c, 0xfb, 0x0c, 0xd1, 0x2b,
-	0x98, 0x38, 0xe1, 0x75, 0x14, 0xdb, 0x9e, 0x77, 0xf1, 0xd5, 0x3d, 0xc3, 0x51, 0x6c, 0xc7, 0xee,
-	0x4a, 0x87, 0x2d, 0xd5, 0x71, 0x8d, 0xab, 0x6b, 0x32, 0x42, 0x87, 0x70, 0xb0, 0x56, 0x6f, 0xcf,
-	0x80, 0x0a, 0x50, 0x8b, 0x3f, 0xbb, 0x6b, 0x0f, 0x8c, 0xd5, 0xc8, 0x47, 0x7a, 0x17, 0xad, 0xab,
-	0xee, 0x2b, 0x4a, 0x1b, 0xfa, 0xff, 0xcd, 0xa0, 0xfe, 0xc4, 0xdf, 0xff, 0x0e, 0x00, 0x00, 0xff,
-	0xff, 0xdb, 0xf5, 0x43, 0x2a, 0xf4, 0x03, 0x00, 0x00,
+func init() {
+	proto.RegisterFile("proto/sharding/p2p/v1/messages.proto", fileDescriptor_messages_8b1affe159ebec96)
+}
+
+var fileDescriptor_messages_8b1affe159ebec96 = []byte{
+	// 459 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x74, 0x92, 0xdf, 0x6e, 0xd3, 0x30,
+	0x14, 0xc6, 0xf1, 0xd6, 0x3f, 0xeb, 0x69, 0x24, 0x22, 0x6b, 0x8c, 0xc0, 0x40, 0x54, 0x85, 0x8b,
+	0xc2, 0x45, 0xaa, 0x15, 0xf1, 0x00, 0x59, 0xa9, 0xc4, 0x44, 0x95, 0x8c, 0x24, 0x13, 0xe2, 0x2a,
+	0xf2, 0x12, 0x2b, 0xb1, 0x68, 0x6d, 0x63, 0x27, 0x91, 0xf6, 0x58, 0xbc, 0x16, 0x4f, 0x81, 0x62,
+	0xa7, 0x63, 0x02, 0x71, 0x97, 0xdf, 0xf7, 0x1d, 0x39, 0xe7, 0xe7, 0x04, 0xde, 0x48, 0x25, 0x6a,
+	0xb1, 0xd4, 0x15, 0x51, 0x05, 0xe3, 0xe5, 0x52, 0xae, 0xe4, 0xb2, 0xbd, 0x58, 0xee, 0xa9, 0xd6,
+	0xa4, 0xa4, 0xda, 0x37, 0x35, 0xf6, 0x68, 0x5d, 0x51, 0x45, 0x9b, 0xbd, 0x7f, 0x18, 0xf4, 0xe5,
+	0x4a, 0xfa, 0xed, 0xc5, 0xfc, 0x27, 0x82, 0xd3, 0xb5, 0xd8, 0xed, 0x48, 0xcd, 0x04, 0xbf, 0x14,
+	0xc5, 0x5d, 0x4c, 0x7f, 0x34, 0x54, 0xd7, 0xf8, 0x19, 0x9c, 0x98, 0xd9, 0x8c, 0x15, 0x1e, 0x9a,
+	0xa1, 0xc5, 0x20, 0x1e, 0x1b, 0xbe, 0x2a, 0xf0, 0x19, 0x8c, 0x24, 0x55, 0x4c, 0x14, 0xde, 0x91,
+	0x29, 0x7a, 0xc2, 0x2f, 0x01, 0xf2, 0xaa, 0xe1, 0xdf, 0x33, 0x25, 0x44, 0xed, 0x1d, 0xcf, 0xd0,
+	0xc2, 0x89, 0x27, 0x26, 0x89, 0x85, 0xa8, 0xf1, 0x5b, 0x70, 0xa5, 0x12, 0x52, 0x68, 0xaa, 0x32,
+	0x52, 0x14, 0x8a, 0x6a, 0xed, 0x0d, 0xcc, 0xd0, 0xe3, 0x43, 0x1e, 0xd8, 0x18, 0xbf, 0x80, 0x89,
+	0x66, 0x25, 0x27, 0x75, 0xa3, 0xa8, 0x37, 0xb4, 0x07, 0xdd, 0x07, 0xf3, 0x2d, 0x3c, 0xf9, 0x6b,
+	0x65, 0x2d, 0x05, 0xd7, 0x14, 0xbf, 0x82, 0x69, 0x45, 0x49, 0x41, 0x55, 0x56, 0x11, 0x5d, 0x99,
+	0xb5, 0x9d, 0x18, 0x6c, 0xf4, 0x89, 0xe8, 0x0a, 0x63, 0x18, 0xdc, 0x8a, 0xe2, 0xce, 0xec, 0xed,
+	0xc4, 0xe6, 0x79, 0xfe, 0x0b, 0xc1, 0x34, 0x55, 0x84, 0x6b, 0x92, 0x77, 0x07, 0xe2, 0x53, 0x18,
+	0x72, 0xc1, 0x73, 0xda, 0x5b, 0x5b, 0xc0, 0xe7, 0x30, 0x29, 0x89, 0xce, 0xa4, 0x62, 0x39, 0xed,
+	0xb5, 0x4f, 0x4a, 0xa2, 0xaf, 0x3b, 0x3e, 0x94, 0x3b, 0xb6, 0x67, 0xd6, 0xdb, 0x96, 0xdb, 0x8e,
+	0x3b, 0x17, 0x45, 0x73, 0x26, 0x19, 0xe5, 0x75, 0xef, 0xfb, 0x27, 0xe8, 0xde, 0xd6, 0x92, 0x5d,
+	0x63, 0x2d, 0x07, 0xb1, 0x85, 0x2e, 0x65, 0x5c, 0x36, 0xb5, 0x37, 0x32, 0xf3, 0x16, 0x70, 0xf0,
+	0xf0, 0x56, 0xc6, 0x33, 0xb4, 0x98, 0xae, 0x5e, 0xfb, 0xff, 0xfb, 0xb2, 0x7e, 0x72, 0x18, 0x7d,
+	0x78, 0x75, 0x1f, 0x60, 0x72, 0x9f, 0x63, 0x07, 0x50, 0xdb, 0x5b, 0xa2, 0xb6, 0x23, 0xd5, 0x9b,
+	0x21, 0xd5, 0x91, 0xee, 0x55, 0x90, 0x7e, 0x97, 0xc1, 0x30, 0x15, 0x92, 0xe5, 0x78, 0x0a, 0xe3,
+	0x9b, 0xf0, 0x73, 0x18, 0x7d, 0x0d, 0xdd, 0x47, 0xf8, 0x39, 0x9c, 0xad, 0xa3, 0xed, 0x36, 0x48,
+	0xaf, 0xa2, 0x30, 0xbb, 0x8c, 0x3e, 0x7e, 0xcb, 0xe2, 0xcd, 0x97, 0x9b, 0x4d, 0x92, 0xba, 0x08,
+	0x9f, 0xc3, 0xd3, 0x7f, 0xba, 0xe4, 0x3a, 0x0a, 0x93, 0x8d, 0x7b, 0x84, 0x5d, 0x70, 0xd2, 0x38,
+	0x08, 0x93, 0x60, 0xdd, 0xd5, 0x89, 0x7b, 0x7c, 0x3b, 0x32, 0xff, 0xe9, 0xfb, 0xdf, 0x01, 0x00,
+	0x00, 0xff, 0xff, 0x1e, 0x58, 0xa9, 0xa1, 0xcf, 0x02, 0x00, 0x00,
 }
--- a/proto/sharding/p2p/v1/messages.proto
+++ b/proto/sharding/p2p/v1/messages.proto
@@ -2,22 +2,11 @@ syntax = "proto3";

 package ethereum.sharding.p2p.v1;

-// TODO: Split the topics into p2p for beacon chain and p2p for sharding.
 enum Topic {
  UNKNOWN = 0;
  COLLATION_BODY_REQUEST = 1;
  COLLATION_BODY_RESPONSE = 2;
  TRANSACTIONS = 3;
-  BEACON_BLOCK_HASH_ANNOUNCE = 4;
-  BEACON_BLOCK_REQUEST = 5;
-  BEACON_BLOCK_REQUEST_BY_SLOT_NUMBER = 6;
-  BEACON_BLOCK_RESPONSE = 7;
-  CRYSTALLIZED_STATE_HASH_ANNOUNCE = 8;
-  CRYSTALLIZED_STATE_REQUEST = 9;
-  CRYSTALLIZED_STATE_RESPONSE = 10;
-  ACTIVE_STATE_HASH_ANNOUNCE = 11;
-  ACTIVE_STATE_REQUEST = 12;
-  ACTIVE_STATE_RESPONSE = 13;
 } 

 message CollationBodyRequest {
--- a/proto/testing/BUILD.bazel
+++ b/proto/testing/BUILD.bazel
@@ -0,0 +1,24 @@
+package(default_testonly = True)
+
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//proto:def.bzl", "go_proto_library")
+
+proto_library(
+    name = "ethereum_testing_proto",
+    srcs = ["test.proto"],
+    visibility = ["//visibility:public"],
+)
+
+go_proto_library(
+    name = "ethereum_testing_go_proto",
+    importpath = "github.com/prysmaticlabs/prysm/proto/testing",
+    proto = ":ethereum_testing_proto",
+    visibility = ["//visibility:public"],
+)
+
+go_library(
+    name = "go_default_library",
+    embed = [":ethereum_testing_go_proto"],
+    importpath = "github.com/prysmaticlabs/prysm/proto/testing",
+    visibility = ["//visibility:public"],
+)
--- a/proto/testing/test.pb.go
+++ b/proto/testing/test.pb.go
@@ -0,0 +1,123 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// source: proto/testing/test.proto
+
+package testing
+
+import proto "github.com/golang/protobuf/proto"
+import fmt "fmt"
+import math "math"
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
+
+type TestMessage struct {
+	Foo                  string   `protobuf:"bytes,1,opt,name=foo" json:"foo,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *TestMessage) Reset()         { *m = TestMessage{} }
+func (m *TestMessage) String() string { return proto.CompactTextString(m) }
+func (*TestMessage) ProtoMessage()    {}
+func (*TestMessage) Descriptor() ([]byte, []int) {
+	return fileDescriptor_test_efe8e22469748e36, []int{0}
+}
+func (m *TestMessage) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_TestMessage.Unmarshal(m, b)
+}
+func (m *TestMessage) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_TestMessage.Marshal(b, m, deterministic)
+}
+func (dst *TestMessage) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_TestMessage.Merge(dst, src)
+}
+func (m *TestMessage) XXX_Size() int {
+	return xxx_messageInfo_TestMessage.Size(m)
+}
+func (m *TestMessage) XXX_DiscardUnknown() {
+	xxx_messageInfo_TestMessage.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_TestMessage proto.InternalMessageInfo
+
+func (m *TestMessage) GetFoo() string {
+	if m != nil {
+		return m.Foo
+	}
+	return ""
+}
+
+type Puzzle struct {
+	Challenge            string   `protobuf:"bytes,1,opt,name=challenge" json:"challenge,omitempty"`
+	Answer               string   `protobuf:"bytes,2,opt,name=answer" json:"answer,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *Puzzle) Reset()         { *m = Puzzle{} }
+func (m *Puzzle) String() string { return proto.CompactTextString(m) }
+func (*Puzzle) ProtoMessage()    {}
+func (*Puzzle) Descriptor() ([]byte, []int) {
+	return fileDescriptor_test_efe8e22469748e36, []int{1}
+}
+func (m *Puzzle) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_Puzzle.Unmarshal(m, b)
+}
+func (m *Puzzle) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_Puzzle.Marshal(b, m, deterministic)
+}
+func (dst *Puzzle) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_Puzzle.Merge(dst, src)
+}
+func (m *Puzzle) XXX_Size() int {
+	return xxx_messageInfo_Puzzle.Size(m)
+}
+func (m *Puzzle) XXX_DiscardUnknown() {
+	xxx_messageInfo_Puzzle.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_Puzzle proto.InternalMessageInfo
+
+func (m *Puzzle) GetChallenge() string {
+	if m != nil {
+		return m.Challenge
+	}
+	return ""
+}
+
+func (m *Puzzle) GetAnswer() string {
+	if m != nil {
+		return m.Answer
+	}
+	return ""
+}
+
+func init() {
+	proto.RegisterType((*TestMessage)(nil), "ethereum.testing.TestMessage")
+	proto.RegisterType((*Puzzle)(nil), "ethereum.testing.Puzzle")
+}
+
+func init() { proto.RegisterFile("proto/testing/test.proto", fileDescriptor_test_efe8e22469748e36) }
+
+var fileDescriptor_test_efe8e22469748e36 = []byte{
+	// 135 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0x92, 0x28, 0x28, 0xca, 0x2f,
+	0xc9, 0xd7, 0x2f, 0x49, 0x2d, 0x2e, 0xc9, 0xcc, 0x4b, 0x07, 0xd3, 0x7a, 0x60, 0x21, 0x21, 0x81,
+	0xd4, 0x92, 0x8c, 0xd4, 0xa2, 0xd4, 0xd2, 0x5c, 0x3d, 0xa8, 0xa4, 0x92, 0x3c, 0x17, 0x77, 0x48,
+	0x6a, 0x71, 0x89, 0x6f, 0x6a, 0x71, 0x71, 0x62, 0x7a, 0xaa, 0x90, 0x00, 0x17, 0x73, 0x5a, 0x7e,
+	0xbe, 0x04, 0xa3, 0x02, 0xa3, 0x06, 0x67, 0x10, 0x88, 0xa9, 0x64, 0xc7, 0xc5, 0x16, 0x50, 0x5a,
+	0x55, 0x95, 0x93, 0x2a, 0x24, 0xc3, 0xc5, 0x99, 0x9c, 0x91, 0x98, 0x93, 0x93, 0x9a, 0x97, 0x9e,
+	0x0a, 0x55, 0x81, 0x10, 0x10, 0x12, 0xe3, 0x62, 0x4b, 0xcc, 0x2b, 0x2e, 0x4f, 0x2d, 0x92, 0x60,
+	0x02, 0x4b, 0x41, 0x79, 0x49, 0x6c, 0x60, 0x9b, 0x8d, 0x01, 0x01, 0x00, 0x00, 0xff, 0xff, 0x7f,
+	0xb3, 0xa3, 0x6b, 0x95, 0x00, 0x00, 0x00,
+}
--- a/proto/testing/test.proto
+++ b/proto/testing/test.proto
@@ -0,0 +1,13 @@
+syntax = "proto3";
+
+package ethereum.testing;
+
+message TestMessage {
+    string foo = 1;
+}
+
+// Used in shared/p2p/feed_example_test.go
+message Puzzle {
+    string challenge = 1;
+    string answer = 2;
+}
--- a/shared/p2p/BUILD.bazel
+++ b/shared/p2p/BUILD.bazel
@@ -7,15 +7,13 @@ go_library(
        "feed.go",
        "message.go",
        "options.go",
+        "p2p.go",
        "peer.go",
        "service.go",
-        "topics.go",
    ],
    importpath = "github.com/prysmaticlabs/prysm/shared/p2p",
    visibility = ["//visibility:public"],
    deps = [
-        "//proto/beacon/p2p/v1:go_default_library",
-        "//proto/sharding/p2p/v1:go_default_library",
        "//shared/iputils:go_default_library",
        "@com_github_ethereum_go_ethereum//event:go_default_library",
        "@com_github_golang_protobuf//proto:go_default_library",
@@ -24,7 +22,6 @@ go_library(
        "@com_github_libp2p_go_libp2p//p2p/discovery:go_default_library",
        "@com_github_libp2p_go_libp2p_crypto//:go_default_library",
        "@com_github_libp2p_go_libp2p_host//:go_default_library",
-        "@com_github_libp2p_go_libp2p_peer//:go_default_library",
        "@com_github_libp2p_go_libp2p_peerstore//:go_default_library",
        "@com_github_multiformats_go_multiaddr//:go_default_library",
        "@com_github_sirupsen_logrus//:go_default_library",
@@ -37,19 +34,22 @@ go_test(
        "feed_example_test.go",
        "feed_test.go",
        "options_test.go",
+        "register_topic_example_test.go",
        "service_test.go",
-        "topics_test.go",
    ],
    embed = [":go_default_library"],
    deps = [
        "//proto/sharding/p2p/v1:go_default_library",
+        "//proto/testing:go_default_library",
        "//shared:go_default_library",
        "@com_github_ethereum_go_ethereum//event:go_default_library",
        "@com_github_golang_protobuf//proto:go_default_library",
        "@com_github_libp2p_go_floodsub//:go_default_library",
-        "@com_github_libp2p_go_libp2p//p2p/host/basic:go_default_library",
+        "@com_github_libp2p_go_floodsub//pb:go_default_library",
+        "@com_github_libp2p_go_libp2p_blankhost//:go_default_library",
        "@com_github_libp2p_go_libp2p_swarm//testing:go_default_library",
        "@com_github_sirupsen_logrus//:go_default_library",
+        "@com_github_sirupsen_logrus//hooks/test:go_default_library",
    ],
 )

--- a/shared/p2p/discovery.go
+++ b/shared/p2p/discovery.go
@@ -5,10 +5,8 @@ import (
 	"time"

 	host "github.com/libp2p/go-libp2p-host"
-	peer "github.com/libp2p/go-libp2p-peer"
 	ps "github.com/libp2p/go-libp2p-peerstore"
 	mdns "github.com/libp2p/go-libp2p/p2p/discovery"
-	shardpb "github.com/prysmaticlabs/prysm/proto/sharding/p2p/v1"
 	"github.com/sirupsen/logrus"
 )

@@ -24,30 +22,20 @@ var mDNSTag = mdns.ServiceTag
 // DNS peer discovery.
 //
 // TODO: add other discovery protocols such as DHT, etc.
-func startDiscovery(ctx context.Context, host host.Host, gsub topicPeerLister) error {
+func startDiscovery(ctx context.Context, host host.Host) error {
 	mdnsService, err := mdns.NewMdnsService(ctx, host, discoveryInterval, mDNSTag)
 	if err != nil {
 		return err
 	}

-	mdnsService.RegisterNotifee(&discovery{ctx, host, gsub})
-
+	mdnsService.RegisterNotifee(&discovery{ctx, host})
 	return nil
 }

-// topicPeerLister has a method to return connected peers on a given topic.
-// This is implemented by floodsub.PubSub.
-type topicPeerLister interface {
-	ListPeers(string) []peer.ID
-}
-
 // Discovery implements mDNS notifee interface.
 type discovery struct {
 	ctx  context.Context
 	host host.Host
-
-	// Required for helper method.
-	gsub topicPeerLister
 }

 // HandlePeerFound registers the peer with the host.
@@ -65,19 +53,4 @@ func (d *discovery) HandlePeerFound(pi ps.PeerInfo) {
 	log.WithFields(logrus.Fields{
 		"peers": d.host.Peerstore().Peers(),
 	}).Debug("Peers are now")
-
-	log.WithFields(logrus.Fields{
-		"peerMap": d.topicPeerMap(),
-	}).Debug("Gsub has peers")
-}
-
-// topicPeerMap helper function for inspecting which peers are available for
-// the p2p topics.
-func (d *discovery) topicPeerMap() map[shardpb.Topic][]peer.ID {
-	m := make(map[shardpb.Topic][]peer.ID)
-	for topic := range topicTypeMapping {
-		peers := d.gsub.ListPeers(topic.String())
-		m[topic] = peers
-	}
-	return m
 }
--- a/shared/p2p/discovery_norace_test.go
+++ b/shared/p2p/discovery_norace_test.go
@@ -5,24 +5,12 @@ import (
 	"testing"
 	"time"

-	floodsub "github.com/libp2p/go-floodsub"
-	peer "github.com/libp2p/go-libp2p-peer"
 	swarmt "github.com/libp2p/go-libp2p-swarm/testing"
 	mdns "github.com/libp2p/go-libp2p/p2p/discovery"
 	bhost "github.com/libp2p/go-libp2p/p2p/host/basic"
 )

 var _ = mdns.Notifee(&discovery{})
-var _ = topicPeerLister(&floodsub.PubSub{})
-
-var _ = topicPeerLister(&fakeTopicPeerLister{})
-
-type fakeTopicPeerLister struct {
-}
-
-func (f *fakeTopicPeerLister) ListPeers(topic string) []peer.ID {
-	return nil
-}

 func expectPeers(t *testing.T, h *bhost.BasicHost, n int) {
 	if len(h.Peerstore().Peers()) != n {
@@ -42,16 +30,14 @@ func TestStartDiscovery_HandlePeerFound(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.TODO())
 	defer cancel()

-	gsub := &fakeTopicPeerLister{}
-
 	a := bhost.New(swarmt.GenSwarm(t, ctx))
-	err := startDiscovery(ctx, a, gsub)
+	err := startDiscovery(ctx, a)
 	if err != nil {
 		t.Errorf("Error when starting discovery: %v", err)
 	}

 	b := bhost.New(swarmt.GenSwarm(t, ctx))
-	err = startDiscovery(ctx, b, gsub)
+	err = startDiscovery(ctx, b)
 	if err != nil {
 		t.Errorf("Error when starting discovery: %v", err)
 	}
--- a/shared/p2p/feed_example_test.go
+++ b/shared/p2p/feed_example_test.go
@@ -1,6 +1,10 @@
 package p2p

-import "fmt"
+import (
+	"fmt"
+
+	pb "github.com/prysmaticlabs/prysm/proto/testing"
+)

 // Feeds can be use to subscribe to any type of message.
 func ExampleServer_Feed() {
@@ -10,12 +14,7 @@ func ExampleServer_Feed() {
 	}

 	// Let's wait for a puzzle from our peers then try to solve it.
-	type Puzzle struct {
-		Challenge string
-		Answer    string
-	}
-
-	feed := s.Feed(Puzzle{})
+	feed := s.Feed(pb.Puzzle{})

 	ch := make(chan Message, 5) // Small buffer size. I don't expect many puzzles.
 	sub := feed.Subscribe(ch)
@@ -26,7 +25,7 @@ func ExampleServer_Feed() {

 	// Wait until we have a puzzle to solve.
 	msg := <-ch
-	puzzle, ok := msg.Data.(Puzzle)
+	puzzle, ok := msg.Data.(*pb.Puzzle)

 	if !ok {
 		panic("Received a message that wasn't a puzzle!")
--- a/shared/p2p/message.go
+++ b/shared/p2p/message.go
@@ -1,9 +1,13 @@
 package p2p

+import (
+	"github.com/golang/protobuf/proto"
+)
+
 // Message represents a message received from an external peer.
 type Message struct {
 	// Peer represents the sender of the message.
 	Peer Peer
 	// Data can be any type of message found in sharding/p2p/proto package.
-	Data interface{}
+	Data proto.Message
 }
--- a/shared/p2p/p2p.go
+++ b/shared/p2p/p2p.go
@@ -0,0 +1,29 @@
+// Package p2p handles peer-to-peer networking for Ethereum 2.0 clients.
+//
+// There are three types of p2p communications.
+//
+// 	- Direct: two peer communication
+// 	- Floodsub: peer broadcasting to all peers
+// 	- Gossipsub: peer broadcasting to localized peers
+//
+// This communication is abstracted through the Feed, Broadcast, and Send.
+//
+// Pub/sub topic has a specific message type that is used for that topic. The
+// mappings for these topics are outlined here: (TODO).
+//
+// Read more about gossipsub at https://github.com/vyzo/gerbil-simsub
+package p2p
+
+import "context"
+
+// Use this file for interfaces only!
+
+// Adapter is used to create middleware.
+//
+// See http://godoc.org/github.com/prysmaticlabs/prysm/shared/p2p#Server.RegisterTopic
+type Adapter func(Handler) Handler
+
+// Handler is a callback used in the adapter/middleware stack chain.
+//
+// See http://godoc.org/github.com/prysmaticlabs/prysm/shared/p2p#Server.RegisterTopic
+type Handler func(context.Context, Message)
--- a/shared/p2p/register_topic_example_test.go
+++ b/shared/p2p/register_topic_example_test.go
@@ -0,0 +1,46 @@
+package p2p_test
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/golang/protobuf/proto"
+	"github.com/prysmaticlabs/prysm/shared/p2p"
+)
+
+// A basic adapter will complete its logic then call next. Some adapters
+// may choose not to call next. For example, in the case of a rate
+// limiter or blacklisting condition.
+func reqLogger(next p2p.Handler) p2p.Handler {
+	return func(ctx context.Context, msg p2p.Message) {
+		fmt.Printf("Received message from %v\n", msg.Peer)
+		next(ctx, msg)
+	}
+}
+
+// Functions can return an adapter in order to capture configuration.
+func adapterWithParams(i int) p2p.Adapter {
+	return func(next p2p.Handler) p2p.Handler {
+		return func(ctx context.Context, msg p2p.Message) {
+			fmt.Printf("Magic number is %d\n", i)
+			i++
+			next(ctx, msg)
+		}
+	}
+}
+
+func ExampleServer_RegisterTopic() {
+	adapters := []p2p.Adapter{reqLogger, adapterWithParams(5)}
+
+	s, _ := p2p.NewServer()
+
+	var topic string
+	var message proto.Message
+
+	s.RegisterTopic(topic, message, adapters...)
+
+	ch := make(chan p2p.Message)
+	sub := s.Subscribe(message, ch)
+	defer sub.Unsubscribe()
+	// TODO: Show more of how the chan is used.
+}
--- a/shared/p2p/service.go
+++ b/shared/p2p/service.go
@@ -1,12 +1,3 @@
-// Package p2p handles peer-to-peer networking for the sharding package.
-//
-// Notes:
-// Gossip sub topics can be identified by their proto message types.
-//
-// 		topic := proto.MessageName(myMsg)
-//
-// Then we can assume that only these message types are broadcast in that
-// gossip subscription.
 package p2p

 import (
@@ -21,7 +12,6 @@ import (
 	floodsub "github.com/libp2p/go-floodsub"
 	libp2p "github.com/libp2p/go-libp2p"
 	host "github.com/libp2p/go-libp2p-host"
-	shardpb "github.com/prysmaticlabs/prysm/proto/sharding/p2p/v1"
 )

 // Sender represents a struct that is able to relay information via p2p.
@@ -32,12 +22,13 @@ type Sender interface {

 // Server is a placeholder for a p2p service. To be designed.
 type Server struct {
-	ctx    context.Context
-	cancel context.CancelFunc
-	mutex  *sync.Mutex
-	feeds  map[reflect.Type]*event.Feed
-	host   host.Host
-	gsub   *floodsub.PubSub
+	ctx          context.Context
+	cancel       context.CancelFunc
+	mutex        *sync.Mutex
+	feeds        map[reflect.Type]*event.Feed
+	host         host.Host
+	gsub         *floodsub.PubSub
+	topicMapping map[reflect.Type]string
 }

 // NewServer creates a new p2p server instance.
@@ -57,30 +48,23 @@ func NewServer() (*Server, error) {
 	}

 	return &Server{
-		ctx:    ctx,
-		cancel: cancel,
-		feeds:  make(map[reflect.Type]*event.Feed),
-		host:   host,
-		gsub:   gsub,
-		mutex:  &sync.Mutex{},
+		ctx:          ctx,
+		cancel:       cancel,
+		feeds:        make(map[reflect.Type]*event.Feed),
+		host:         host,
+		gsub:         gsub,
+		mutex:        &sync.Mutex{},
+		topicMapping: make(map[reflect.Type]string),
 	}, nil
 }

 // Start the main routine for an p2p server.
 func (s *Server) Start() {
 	log.Info("Starting service")
-	if err := startDiscovery(s.ctx, s.host, s.gsub); err != nil {
+	if err := startDiscovery(s.ctx, s.host); err != nil {
 		log.Errorf("Could not start p2p discovery! %v", err)
 		return
 	}
-
-	// Subscribe to all topics.
-	for topic, msgType := range topicTypeMapping {
-		log.WithFields(logrus.Fields{
-			"topic": topic,
-		}).Debug("Subscribing to topic")
-		go s.subscribeToTopic(topic, msgType)
-	}
 }

 // Stop the main p2p loop.
@@ -91,6 +75,82 @@ func (s *Server) Stop() error {
 	return nil
 }

+// RegisterTopic with a message and the adapter stack for the given topic. The
+// message type provided will be feed selector for emitting messages received
+// on a given topic.
+//
+// The topics can originate from multiple sources. In other words, messages on
+// TopicA may come from direct peer communication or a pub/sub channel.
+func (s *Server) RegisterTopic(topic string, message interface{}, adapters ...Adapter) {
+	msgType := reflect.TypeOf(message)
+	log.WithFields(logrus.Fields{
+		"topic": topic,
+	}).Debug("Subscribing to topic")
+
+	s.topicMapping[msgType] = topic
+
+	sub, err := s.gsub.Subscribe(topic)
+	if err != nil {
+		log.Errorf("Failed to subscribe to topic: %v", err)
+		return
+	}
+	feed := s.Feed(msgType)
+
+	// Reverse adapter order
+	for i := len(adapters)/2 - 1; i >= 0; i-- {
+		opp := len(adapters) - 1 - i
+		adapters[i], adapters[opp] = adapters[opp], adapters[i]
+	}
+
+	go func() {
+		defer sub.Cancel()
+		for {
+			msg, err := sub.Next(s.ctx)
+
+			if s.ctx.Err() != nil {
+				log.WithError(s.ctx.Err()).Debug("Context error")
+				return
+			}
+
+			if err != nil {
+				log.Errorf("Failed to get next message: %v", err)
+				return
+			}
+
+			var h Handler = func(ctx context.Context, pMsg Message) {
+				s.emit(feed, msg, msgType)
+			}
+
+			pMsg := Message{}
+
+			for _, adapter := range adapters {
+				h = adapter(h)
+			}
+
+			h(s.ctx, pMsg)
+		}
+	}()
+
+}
+
+func (s *Server) emit(feed *event.Feed, msg *floodsub.Message, msgType reflect.Type) {
+	d, ok := reflect.New(msgType).Interface().(proto.Message)
+	if !ok {
+		log.Error("Received message is not a protobuf message")
+		return
+	}
+	if err := proto.Unmarshal(msg.Data, d); err != nil {
+		log.Errorf("Failed to decode data: %v", err)
+		return
+	}
+
+	i := feed.Send(Message{Data: d})
+	log.WithFields(logrus.Fields{
+		"numSubs": i,
+	}).Debug("Sent a request to subs")
+
+}
+
 // Subscribe returns a subscription to a feed of msg's Type and adds the channels to the feed.
 func (s *Server) Subscribe(msg interface{}, channel interface{}) event.Subscription {
 	return s.Feed(msg).Subscribe(channel)
@@ -112,12 +172,12 @@ func (s *Server) Send(msg interface{}, peer Peer) {
 // Broadcast a message to the world.
 func (s *Server) Broadcast(msg interface{}) {
 	// TODO: https://github.com/prysmaticlabs/prysm/issues/176
-	topic := topic(msg)
+	topic := s.topicMapping[reflect.TypeOf(msg)]
 	log.WithFields(logrus.Fields{
 		"topic": topic,
-	}).Debugf("Broadcasting msg %T", msg)
+	}).Debugf("Broadcasting msg %s", msg)

-	if topic == shardpb.Topic_UNKNOWN {
+	if topic == "" {
 		log.Warnf("Topic is unknown for message type %T. %v", msg, msg)
 	}

@@ -133,47 +193,7 @@ func (s *Server) Broadcast(msg interface{}) {
 		log.Errorf("Failed to marshal data for broadcast: %v", err)
 		return
 	}
-	if err := s.gsub.Publish(topic.String(), b); err != nil {
+	if err := s.gsub.Publish(topic, b); err != nil {
 		log.Errorf("Failed to publish to gossipsub topic: %v", err)
 	}
 }
-
-func (s *Server) subscribeToTopic(topic shardpb.Topic, msgType reflect.Type) {
-	sub, err := s.gsub.Subscribe(topic.String())
-	if err != nil {
-		log.Errorf("Failed to subscribe to topic: %v", err)
-		return
-	}
-	defer sub.Cancel()
-	feed := s.Feed(msgType)
-
-	for {
-		msg, err := sub.Next(s.ctx)
-
-		if s.ctx.Err() != nil {
-			return // Context closed or something.
-		}
-		if err != nil {
-			log.Errorf("Failed to get next message: %v", err)
-			return
-		}
-
-		// TODO: reflect.Value.Interface() can panic so we should capture that
-		// panic so the server doesn't crash.
-		d, ok := reflect.New(msgType).Interface().(proto.Message)
-		if !ok {
-			log.Error("Received message is not a protobuf message")
-			continue
-		}
-		err = proto.Unmarshal(msg.Data, d)
-		if err != nil {
-			log.Errorf("Failed to decode data: %v", err)
-			continue
-		}
-
-		i := feed.Send(Message{Data: d})
-		log.WithFields(logrus.Fields{
-			"numSubs": i,
-		}).Debug("Sent a request to subs")
-	}
-}
--- a/shared/p2p/service_test.go
+++ b/shared/p2p/service_test.go
@@ -4,19 +4,22 @@ import (
 	"context"
 	"io/ioutil"
 	"reflect"
+	"strings"
 	"sync"
 	"testing"
 	"time"

 	"github.com/ethereum/go-ethereum/event"
 	"github.com/golang/protobuf/proto"
-	"github.com/prysmaticlabs/prysm/shared"
-
 	floodsub "github.com/libp2p/go-floodsub"
+	floodsubPb "github.com/libp2p/go-floodsub/pb"
+	bhost "github.com/libp2p/go-libp2p-blankhost"
 	swarmt "github.com/libp2p/go-libp2p-swarm/testing"
-	bhost "github.com/libp2p/go-libp2p/p2p/host/basic"
 	shardpb "github.com/prysmaticlabs/prysm/proto/sharding/p2p/v1"
+	testpb "github.com/prysmaticlabs/prysm/proto/testing"
+	"github.com/prysmaticlabs/prysm/shared"
 	"github.com/sirupsen/logrus"
+	logTest "github.com/sirupsen/logrus/hooks/test"
 )

 // Ensure that server implements service.
@@ -39,10 +42,36 @@ func TestBroadcast(t *testing.T) {
 	// TODO: test that topic was published
 }

+func TestEmitFailsNonProtobuf(t *testing.T) {
+	s, _ := NewServer()
+	hook := logTest.NewGlobal()
+	s.emit(nil /*feed*/, nil /*msg*/, reflect.TypeOf(""))
+	want := "Received message is not a protobuf message"
+	if hook.LastEntry().Message != want {
+		t.Errorf("Expected log to contain %s. Got = %s", want, hook.LastEntry().Message)
+	}
+}
+
+func TestEmitFailsUnmarshal(t *testing.T) {
+	s, _ := NewServer()
+	hook := logTest.NewGlobal()
+	msg := &floodsub.Message{
+		&floodsubPb.Message{
+			Data: []byte("bogus"),
+		},
+	}
+
+	s.emit(nil /*feed*/, msg, reflect.TypeOf(testpb.TestMessage{}))
+	want := "Failed to decode data:"
+	if !strings.Contains(hook.LastEntry().Message, want) {
+		t.Errorf("Expected log to contain %s. Got = %s", want, hook.LastEntry().Message)
+	}
+}
+
 func TestSubscribeToTopic(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.TODO(), 1*time.Second)
 	defer cancel()
-	h := bhost.New(swarmt.GenSwarm(t, ctx))
+	h := bhost.NewBlankHost(swarmt.GenSwarm(t, ctx))

 	gsub, err := floodsub.NewFloodSub(ctx, h)
 	if err != nil {
@@ -50,11 +79,12 @@ func TestSubscribeToTopic(t *testing.T) {
 	}

 	s := Server{
-		ctx:   ctx,
-		gsub:  gsub,
-		host:  h,
-		feeds: make(map[reflect.Type]*event.Feed),
-		mutex: &sync.Mutex{},
+		ctx:          ctx,
+		gsub:         gsub,
+		host:         h,
+		feeds:        make(map[reflect.Type]*event.Feed),
+		mutex:        &sync.Mutex{},
+		topicMapping: make(map[reflect.Type]string),
 	}

 	feed := s.Feed(shardpb.CollationBodyRequest{})
@@ -68,7 +98,7 @@ func TestSubscribeToTopic(t *testing.T) {
 func TestSubscribe(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.TODO(), 1*time.Second)
 	defer cancel()
-	h := bhost.New(swarmt.GenSwarm(t, ctx))
+	h := bhost.NewBlankHost(swarmt.GenSwarm(t, ctx))

 	gsub, err := floodsub.NewFloodSub(ctx, h)
 	if err != nil {
@@ -76,11 +106,12 @@ func TestSubscribe(t *testing.T) {
 	}

 	s := Server{
-		ctx:   ctx,
-		gsub:  gsub,
-		host:  h,
-		feeds: make(map[reflect.Type]*event.Feed),
-		mutex: &sync.Mutex{},
+		ctx:          ctx,
+		gsub:         gsub,
+		host:         h,
+		feeds:        make(map[reflect.Type]*event.Feed),
+		mutex:        &sync.Mutex{},
+		topicMapping: make(map[reflect.Type]string),
 	}

 	ch := make(chan Message)
@@ -92,8 +123,8 @@ func TestSubscribe(t *testing.T) {

 func testSubscribe(ctx context.Context, t *testing.T, s Server, gsub *floodsub.PubSub, ch chan Message) {
 	topic := shardpb.Topic_COLLATION_BODY_REQUEST
-	msgType := topicTypeMapping[topic]
-	go s.subscribeToTopic(topic, msgType)
+
+	go s.RegisterTopic(topic.String(), shardpb.CollationBodyRequest{})

 	// Short delay to let goroutine add subscription.
 	time.Sleep(time.Millisecond * 10)
@@ -132,3 +163,106 @@ func testSubscribe(ctx context.Context, t *testing.T, s Server, gsub *floodsub.P
 		t.Error("Context timed out before a message was received!")
 	}
 }
+
+func TestRegisterTopic_WithoutAdapters(t *testing.T) {
+	s, err := NewServer()
+	if err != nil {
+		t.Fatalf("Failed to create new server: %v", err)
+	}
+	topic := "test_topic"
+	testMessage := testpb.TestMessage{Foo: "bar"}
+
+	s.RegisterTopic(topic, testpb.TestMessage{})
+
+	ch := make(chan Message)
+	sub := s.Subscribe(testMessage, ch)
+	defer sub.Unsubscribe()
+
+	wait := make(chan struct{})
+	go func() {
+		defer close(wait)
+		<-ch
+	}()
+
+	if err := simulateIncomingMessage(t, s, topic, []byte{}); err != nil {
+		t.Errorf("Failed to send to topic %s", topic)
+	}
+
+	select {
+	case <-wait:
+		return // OK
+	case <-time.After(1 * time.Second):
+		t.Fatal("TestMessage not received within 1 seconds")
+	}
+}
+
+func TestRegisterTopic_WithAdapers(t *testing.T) {
+	s, err := NewServer()
+	if err != nil {
+		t.Fatalf("Failed to create new server: %v", err)
+	}
+	topic := "test_topic"
+	testMessage := testpb.TestMessage{Foo: "bar"}
+
+	i := 0
+	var testAdapter Adapter = func(next Handler) Handler {
+		return func(ctx context.Context, msg Message) {
+			i++
+			next(ctx, msg)
+		}
+	}
+
+	adapters := []Adapter{
+		testAdapter,
+		testAdapter,
+		testAdapter,
+		testAdapter,
+		testAdapter,
+	}
+
+	s.RegisterTopic(topic, testpb.TestMessage{}, adapters...)
+
+	ch := make(chan Message)
+	sub := s.Subscribe(testMessage, ch)
+	defer sub.Unsubscribe()
+
+	wait := make(chan struct{})
+	go func() {
+		defer close(wait)
+		<-ch
+	}()
+
+	if err := simulateIncomingMessage(t, s, topic, []byte{}); err != nil {
+		t.Errorf("Failed to send to topic %s", topic)
+	}
+
+	select {
+	case <-wait:
+		if i != 5 {
+			t.Errorf("Expected testAdapter to increment i to 5, but was %d", i)
+		}
+		return // OK
+	case <-time.After(1 * time.Second):
+		t.Fatal("TestMessage not received within 1 seconds")
+	}
+}
+
+func simulateIncomingMessage(t *testing.T, s *Server, topic string, b []byte) error {
+	ctx := context.Background()
+	h := bhost.NewBlankHost(swarmt.GenSwarm(t, ctx))
+
+	gsub, err := floodsub.NewFloodSub(ctx, h)
+	if err != nil {
+		return err
+	}
+
+	pinfo := h.Peerstore().PeerInfo(h.ID())
+	if err = s.host.Connect(ctx, pinfo); err != nil {
+		return err
+	}
+
+	// Short timeout to allow libp2p to handle peer connection.
+	time.Sleep(time.Millisecond * 10)
+
+	return gsub.Publish(topic, b)
+}
--- a/shared/p2p/topics.go
+++ b/shared/p2p/topics.go
@@ -1,48 +0,0 @@
-package p2p
-
-import (
-	"reflect"
-
-	beaconpb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
-	shardpb "github.com/prysmaticlabs/prysm/proto/sharding/p2p/v1"
-)
-
-// Mapping of message topic enums to protobuf types.
-var topicTypeMapping = map[shardpb.Topic]reflect.Type{
-	shardpb.Topic_BEACON_BLOCK_HASH_ANNOUNCE:          reflect.TypeOf(beaconpb.BeaconBlockHashAnnounce{}),
-	shardpb.Topic_BEACON_BLOCK_REQUEST:                reflect.TypeOf(beaconpb.BeaconBlockRequest{}),
-	shardpb.Topic_BEACON_BLOCK_REQUEST_BY_SLOT_NUMBER: reflect.TypeOf(beaconpb.BeaconBlockRequestBySlotNumber{}),
-	shardpb.Topic_BEACON_BLOCK_RESPONSE:               reflect.TypeOf(beaconpb.BeaconBlockResponse{}),
-	shardpb.Topic_COLLATION_BODY_REQUEST:              reflect.TypeOf(shardpb.CollationBodyRequest{}),
-	shardpb.Topic_COLLATION_BODY_RESPONSE:             reflect.TypeOf(shardpb.CollationBodyResponse{}),
-	shardpb.Topic_TRANSACTIONS:                        reflect.TypeOf(shardpb.Transaction{}),
-	shardpb.Topic_CRYSTALLIZED_STATE_HASH_ANNOUNCE:    reflect.TypeOf(beaconpb.CrystallizedStateHashAnnounce{}),
-	shardpb.Topic_CRYSTALLIZED_STATE_REQUEST:          reflect.TypeOf(beaconpb.CrystallizedStateRequest{}),
-	shardpb.Topic_CRYSTALLIZED_STATE_RESPONSE:         reflect.TypeOf(beaconpb.CrystallizedStateResponse{}),
-	shardpb.Topic_ACTIVE_STATE_HASH_ANNOUNCE:          reflect.TypeOf(beaconpb.ActiveStateHashAnnounce{}),
-	shardpb.Topic_ACTIVE_STATE_REQUEST:                reflect.TypeOf(beaconpb.ActiveStateRequest{}),
-	shardpb.Topic_ACTIVE_STATE_RESPONSE:               reflect.TypeOf(beaconpb.ActiveStateResponse{}),
-}
-
-// Mapping of message types to topic enums.
-var typeTopicMapping = reverseMapping(topicTypeMapping)
-
-// ReverseMapping from K,V to V,K
-func reverseMapping(m map[shardpb.Topic]reflect.Type) map[reflect.Type]shardpb.Topic {
-	n := make(map[reflect.Type]shardpb.Topic)
-	for k, v := range m {
-		n[v] = k
-	}
-	return n
-}
-
-// These functions return the given topic for a given interface. This is the preferred
-// way to resolve a topic from an value. The msg could be a pointer or value
-// argument to resolve to the correct topic.
-func topic(msg interface{}) shardpb.Topic {
-	msgType := reflect.TypeOf(msg)
-	if msgType.Kind() == reflect.Ptr {
-		msgType = reflect.Indirect(reflect.ValueOf(msg)).Type()
-	}
-	return typeTopicMapping[msgType]
-}
--- a/shared/p2p/topics_test.go
+++ b/shared/p2p/topics_test.go
@@ -1,62 +0,0 @@
-package p2p
-
-import (
-	"reflect"
-	"testing"
-
-	shardpb "github.com/prysmaticlabs/prysm/proto/sharding/p2p/v1"
-)
-
-type testStruct struct{}
-
-func TestReverseMapping(t *testing.T) {
-	tests := []struct {
-		input map[shardpb.Topic]reflect.Type
-		want  map[reflect.Type]shardpb.Topic
-	}{
-		{
-			input: map[shardpb.Topic]reflect.Type{
-				shardpb.Topic_UNKNOWN: reflect.TypeOf(testStruct{}),
-			},
-			want: map[reflect.Type]shardpb.Topic{
-				reflect.TypeOf(testStruct{}): shardpb.Topic_UNKNOWN,
-			},
-		},
-	}
-
-	for _, tt := range tests {
-		got := reverseMapping(tt.input)
-		if !reflect.DeepEqual(got, tt.want) {
-			t.Errorf("reverseMapping(%+v) = %+v. Wanted %+v", tt.input, got, tt.want)
-		}
-	}
-}
-
-func TestTopic(t *testing.T) {
-	type CustomStruct struct{}
-
-	tests := []struct {
-		input interface{}
-		want  shardpb.Topic
-	}{
-		{
-			input: shardpb.CollationBodyRequest{},
-			want:  shardpb.Topic_COLLATION_BODY_REQUEST,
-		},
-		{
-			input: &shardpb.CollationBodyRequest{},
-			want:  shardpb.Topic_COLLATION_BODY_REQUEST,
-		},
-		{
-			input: CustomStruct{},
-			want:  shardpb.Topic_UNKNOWN,
-		},
-	}
-
-	for _, tt := range tests {
-		got := topic(tt.input)
-		if got != tt.want {
-			t.Errorf("topic(%T) = %v. wanted %v", tt.input, got, tt.want)
-		}
-	}
-}
--- a/validator/node/BUILD.bazel
+++ b/validator/node/BUILD.bazel
@@ -1,11 +1,22 @@
 load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")

+go_test(
+    name = "go_default_test",
+    srcs = ["node_test.go"],
+    embed = [":go_default_library"],
+    deps = ["@com_github_urfave_cli//:go_default_library"],
+)
+
 go_library(
    name = "go_default_library",
-    srcs = ["node.go"],
+    srcs = [
+        "node.go",
+        "p2p_config.go",
+    ],
    importpath = "github.com/prysmaticlabs/prysm/validator/node",
    visibility = ["//validator:__subpackages__"],
    deps = [
+        "//proto/sharding/p2p/v1:go_default_library",
        "//shared:go_default_library",
        "//shared/cmd:go_default_library",
        "//shared/database:go_default_library",
@@ -21,10 +32,3 @@ go_library(
        "@com_github_urfave_cli//:go_default_library",
    ],
 )
-
-go_test(
-    name = "go_default_test",
-    srcs = ["node_test.go"],
-    embed = [":go_default_library"],
-    deps = ["@com_github_urfave_cli//:go_default_library"],
-)
--- a/validator/node/node.go
+++ b/validator/node/node.go
@@ -137,7 +137,7 @@ func (s *ShardEthereum) startDB(ctx *cli.Context) error {

 // registerP2P attaches a p2p server to the ShardEthereum instance.
 func (s *ShardEthereum) registerP2P() error {
-	shardp2p, err := p2p.NewServer()
+	shardp2p, err := configureP2P()
 	if err != nil {
 		return fmt.Errorf("could not register shardp2p service: %v", err)
 	}
--- a/validator/node/p2p_config.go
+++ b/validator/node/p2p_config.go
@@ -0,0 +1,28 @@
+package node
+
+import (
+	"github.com/prysmaticlabs/prysm/shared/p2p"
+
+	pb "github.com/prysmaticlabs/prysm/proto/sharding/p2p/v1"
+)
+
+var topicMappings = map[pb.Topic]interface{}{
+	pb.Topic_COLLATION_BODY_REQUEST:  pb.CollationBodyRequest{},
+	pb.Topic_COLLATION_BODY_RESPONSE: pb.CollationBodyResponse{},
+	pb.Topic_TRANSACTIONS:            pb.Transaction{},
+}
+
+func configureP2P() (*p2p.Server, error) {
+	s, err := p2p.NewServer()
+	if err != nil {
+		return nil, err
+	}
+
+	// TODO(437, 438): Define default adapters for logging, monitoring, etc.
+	var adapters []p2p.Adapter
+	for k, v := range topicMappings {
+		s.RegisterTopic(k.String(), v, adapters...)
+	}
+
+	return s, nil
+}