blockchain: make SubscribeFor* channel arguments send-only

There is no need to accept rw channel. Strengthening the type to
send-only will allow the caller to ensure control of reading from the
provided channel.

Closes #2885

Signed-off-by: Ekaterina Pavlova <ekt@morphbits.io>

internal/fakechain/fakechain.go

@@ -24,7 +24,7 @@ import (
 type FakeChain struct {
  config.Blockchain
  *mempool.Pool
- blocksCh []chan *block.Block
+ blocksCh []chan<- *block.Block
  Blockheight atomic.Uint32
  PoolTxF func(*transaction.Transaction) error
  poolTxWithData func(*transaction.Transaction, any, *mempool.Pool) error
@@ -346,7 +346,7 @@ func (chain *FakeChain) PoolTx(tx *transaction.Transaction, _ ...*mempool.Pool)
 }
 
 // SubscribeForBlocks implements the Blockchainer interface.
-func (chain *FakeChain) SubscribeForBlocks(ch chan *block.Block) {
+func (chain *FakeChain) SubscribeForBlocks(ch chan<- *block.Block) {
  chain.blocksCh = append(chain.blocksCh, ch)
 }
 
@@ -379,7 +379,7 @@ func (chain *FakeChain) VerifyWitness(util.Uint160, hash.Hashable, *transaction.
 }
 
 // UnsubscribeFromBlocks implements the Blockchainer interface.
-func (chain *FakeChain) UnsubscribeFromBlocks(ch chan *block.Block) {
+func (chain *FakeChain) UnsubscribeFromBlocks(ch chan<- *block.Block) {
  for i, c := range chain.blocksCh {
  if c == ch {
  if i < len(chain.blocksCh) {

pkg/consensus/consensus.go

@@ -50,8 +50,8 @@ type Ledger interface {
  GetTransaction(util.Uint256) (*transaction.Transaction, uint32, error)
  ComputeNextBlockValidators() []*keys.PublicKey
  PoolTx(t *transaction.Transaction, pools ...*mempool.Pool) error
- SubscribeForBlocks(ch chan *coreb.Block)
- UnsubscribeFromBlocks(ch chan *coreb.Block)
+ SubscribeForBlocks(ch chan<- *coreb.Block)
+ UnsubscribeFromBlocks(ch chan<- *coreb.Block)
  GetBaseExecFee() int64
  CalculateAttributesFee(tx *transaction.Transaction) int64
  interop.Ledger

pkg/core/blockchain.go

@@ -1327,42 +1327,42 @@ func (bc *Blockchain) notificationDispatcher() {
  // These are just sets of subscribers, though modelled as maps
  // for ease of management (not a lot of subscriptions is really
  // expected, but maps are convenient for adding/deleting elements).
- blockFeed = make(map[chan *block.Block]bool)
- headerFeed = make(map[chan *block.Header]bool)
- txFeed = make(map[chan *transaction.Transaction]bool)
- notificationFeed = make(map[chan *state.ContainedNotificationEvent]bool)
- executionFeed = make(map[chan *state.AppExecResult]bool)
+ blockFeed = make(map[chan<- *block.Block]bool)
+ headerFeed = make(map[chan<- *block.Header]bool)
+ txFeed = make(map[chan<- *transaction.Transaction]bool)
+ notificationFeed = make(map[chan<- *state.ContainedNotificationEvent]bool)
+ executionFeed = make(map[chan<- *state.AppExecResult]bool)
  )
  for {
  select {
  case <-bc.stopCh:
  return
  case sub := <-bc.subCh:
  switch ch := sub.(type) {
- case chan *block.Header:
+ case chan<- *block.Header:
  headerFeed[ch] = true
- case chan *block.Block:
+ case chan<- *block.Block:
  blockFeed[ch] = true
- case chan *transaction.Transaction:
+ case chan<- *transaction.Transaction:
  txFeed[ch] = true
- case chan *state.ContainedNotificationEvent:
+ case chan<- *state.ContainedNotificationEvent:
  notificationFeed[ch] = true
- case chan *state.AppExecResult:
+ case chan<- *state.AppExecResult:
  executionFeed[ch] = true
  default:
  panic(fmt.Sprintf("bad subscription: %T", sub))
  }
  case unsub := <-bc.unsubCh:
  switch ch := unsub.(type) {
- case chan *block.Header:
+ case chan<- *block.Header:
  delete(headerFeed, ch)
- case chan *block.Block:
+ case chan<- *block.Block:
  delete(blockFeed, ch)
- case chan *transaction.Transaction:
+ case chan<- *transaction.Transaction:
  delete(txFeed, ch)
- case chan *state.ContainedNotificationEvent:
+ case chan<- *state.ContainedNotificationEvent:
  delete(notificationFeed, ch)
- case chan *state.AppExecResult:
+ case chan<- *state.AppExecResult:
  delete(executionFeed, ch)
  default:
  panic(fmt.Sprintf("bad unsubscription: %T", unsub))
@@ -2285,7 +2285,7 @@ func (bc *Blockchain) GetConfig() config.Blockchain {
 // Make sure it's read from regularly as not reading these events might affect
 // other Blockchain functions. Make sure you're not changing the received blocks,
 // as it may affect the functionality of Blockchain and other subscribers.
-func (bc *Blockchain) SubscribeForBlocks(ch chan *block.Block) {
+func (bc *Blockchain) SubscribeForBlocks(ch chan<- *block.Block) {
  bc.subCh <- ch
 }
 
@@ -2295,7 +2295,7 @@ func (bc *Blockchain) SubscribeForBlocks(ch chan *block.Block) {
 // affect other Blockchain functions. Make sure you're not changing the received
 // headers, as it may affect the functionality of Blockchain and other
 // subscribers.
-func (bc *Blockchain) SubscribeForHeadersOfAddedBlocks(ch chan *block.Header) {
+func (bc *Blockchain) SubscribeForHeadersOfAddedBlocks(ch chan<- *block.Header) {
  bc.subCh <- ch
 }
 
@@ -2305,7 +2305,7 @@ func (bc *Blockchain) SubscribeForHeadersOfAddedBlocks(ch chan *block.Header) {
 // as not reading these events might affect other Blockchain functions. Make sure
 // you're not changing the received transactions, as it may affect the
 // functionality of Blockchain and other subscribers.
-func (bc *Blockchain) SubscribeForTransactions(ch chan *transaction.Transaction) {
+func (bc *Blockchain) SubscribeForTransactions(ch chan<- *transaction.Transaction) {
  bc.subCh <- ch
 }
 
@@ -2317,7 +2317,7 @@ func (bc *Blockchain) SubscribeForTransactions(ch chan *transaction.Transaction)
 // read from regularly as not reading these events might affect other Blockchain
 // functions. Make sure you're not changing the received notification events, as
 // it may affect the functionality of Blockchain and other subscribers.
-func (bc *Blockchain) SubscribeForNotifications(ch chan *state.ContainedNotificationEvent) {
+func (bc *Blockchain) SubscribeForNotifications(ch chan<- *state.ContainedNotificationEvent) {
  bc.subCh <- ch
 }
 
@@ -2327,18 +2327,17 @@ func (bc *Blockchain) SubscribeForNotifications(ch chan *state.ContainedNotifica
 // reading these events might affect other Blockchain functions. Make sure you're
 // not changing the received execution results, as it may affect the
 // functionality of Blockchain and other subscribers.
-func (bc *Blockchain) SubscribeForExecutions(ch chan *state.AppExecResult) {
+func (bc *Blockchain) SubscribeForExecutions(ch chan<- *state.AppExecResult) {
  bc.subCh <- ch
 }
 
 // UnsubscribeFromBlocks unsubscribes given channel from new block notifications,
 // you can close it afterwards. Passing non-subscribed channel is a no-op, but
 // the method can read from this channel (discarding any read data).
-func (bc *Blockchain) UnsubscribeFromBlocks(ch chan *block.Block) {
+func (bc *Blockchain) UnsubscribeFromBlocks(ch chan<- *block.Block) {
 unsubloop:
  for {
  select {
- case <-ch:
  case bc.unsubCh <- ch:
  break unsubloop
  }
@@ -2349,11 +2348,10 @@ unsubloop:
 // block's header notifications, you can close it afterwards. Passing
 // non-subscribed channel is a no-op, but the method can read from this
 // channel (discarding any read data).
-func (bc *Blockchain) UnsubscribeFromHeadersOfAddedBlocks(ch chan *block.Header) {
+func (bc *Blockchain) UnsubscribeFromHeadersOfAddedBlocks(ch chan<- *block.Header) {
 unsubloop:
  for {
  select {
- case <-ch:
  case bc.unsubCh <- ch:
  break unsubloop
  }
@@ -2363,11 +2361,10 @@ unsubloop:
 // UnsubscribeFromTransactions unsubscribes given channel from new transaction
 // notifications, you can close it afterwards. Passing non-subscribed channel is
 // a no-op, but the method can read from this channel (discarding any read data).
-func (bc *Blockchain) UnsubscribeFromTransactions(ch chan *transaction.Transaction) {
+func (bc *Blockchain) UnsubscribeFromTransactions(ch chan<- *transaction.Transaction) {
 unsubloop:
  for {
  select {
- case <-ch:
  case bc.unsubCh <- ch:
  break unsubloop
  }
@@ -2378,11 +2375,10 @@ unsubloop:
 // execution-generated notifications, you can close it afterwards. Passing
 // non-subscribed channel is a no-op, but the method can read from this channel
 // (discarding any read data).
-func (bc *Blockchain) UnsubscribeFromNotifications(ch chan *state.ContainedNotificationEvent) {
+func (bc *Blockchain) UnsubscribeFromNotifications(ch chan<- *state.ContainedNotificationEvent) {
 unsubloop:
  for {
  select {
- case <-ch:
  case bc.unsubCh <- ch:
  break unsubloop
  }
@@ -2392,11 +2388,10 @@ unsubloop:
 // UnsubscribeFromExecutions unsubscribes given channel from new execution
 // notifications, you can close it afterwards. Passing non-subscribed channel is
 // a no-op, but the method can read from this channel (discarding any read data).
-func (bc *Blockchain) UnsubscribeFromExecutions(ch chan *state.AppExecResult) {
+func (bc *Blockchain) UnsubscribeFromExecutions(ch chan<- *state.AppExecResult) {
 unsubloop:
  for {
  select {
- case <-ch:
  case bc.unsubCh <- ch:
  break unsubloop
  }

pkg/network/server.go

@@ -74,8 +74,8 @@ type (
  PoolTx(t *transaction.Transaction, pools ...*mempool.Pool) error
  PoolTxWithData(t *transaction.Transaction, data any, mp *mempool.Pool, feer mempool.Feer, verificationFunction func(t *transaction.Transaction, data any) error) error
  RegisterPostBlock(f func(func(*transaction.Transaction, *mempool.Pool, bool) bool, *mempool.Pool, *block.Block))
- SubscribeForBlocks(ch chan *block.Block)
- UnsubscribeFromBlocks(ch chan *block.Block)
+ SubscribeForBlocks(ch chan<- *block.Block)
+ UnsubscribeFromBlocks(ch chan<- *block.Block)
  }
 
  // Service is a service abstraction (oracle, state root, consensus, etc).

pkg/services/notary/notary.go

@@ -33,8 +33,8 @@ type (
  BlockHeight() uint32
  GetMaxVerificationGAS() int64
  GetNotaryContractScriptHash() util.Uint160
- SubscribeForBlocks(ch chan *block.Block)
- UnsubscribeFromBlocks(ch chan *block.Block)
+ SubscribeForBlocks(ch chan<- *block.Block)
+ UnsubscribeFromBlocks(ch chan<- *block.Block)
  VerifyWitness(util.Uint160, hash.Hashable, *transaction.Witness, int64) (int64, error)
  }
 

pkg/services/rpcsrv/server.go

@@ -99,16 +99,16 @@ type (
  HeaderHeight() uint32
  InitVerificationContext(ic *interop.Context, hash util.Uint160, witness *transaction.Witness) error
  P2PSigExtensionsEnabled() bool
- SubscribeForBlocks(ch chan *block.Block)
- SubscribeForHeadersOfAddedBlocks(ch chan *block.Header)
- SubscribeForExecutions(ch chan *state.AppExecResult)
- SubscribeForNotifications(ch chan *state.ContainedNotificationEvent)
- SubscribeForTransactions(ch chan *transaction.Transaction)
- UnsubscribeFromBlocks(ch chan *block.Block)
- UnsubscribeFromHeadersOfAddedBlocks(ch chan *block.Header)
- UnsubscribeFromExecutions(ch chan *state.AppExecResult)
- UnsubscribeFromNotifications(ch chan *state.ContainedNotificationEvent)
- UnsubscribeFromTransactions(ch chan *transaction.Transaction)
+ SubscribeForBlocks(ch chan<- *block.Block)
+ SubscribeForHeadersOfAddedBlocks(ch chan<- *block.Header)
+ SubscribeForExecutions(ch chan<- *state.AppExecResult)
+ SubscribeForNotifications(ch chan<- *state.ContainedNotificationEvent)
+ SubscribeForTransactions(ch chan<- *transaction.Transaction)
+ UnsubscribeFromBlocks(ch chan<- *block.Block)
+ UnsubscribeFromHeadersOfAddedBlocks(ch chan<- *block.Header)
+ UnsubscribeFromExecutions(ch chan<- *state.AppExecResult)
+ UnsubscribeFromNotifications(ch chan<- *state.ContainedNotificationEvent)
+ UnsubscribeFromTransactions(ch chan<- *transaction.Transaction)
  VerifyTx(*transaction.Transaction) error
  VerifyWitness(util.Uint160, hash.Hashable, *transaction.Witness, int64) (int64, error)
  mempool.Feer // fee interface

pkg/services/stateroot/service.go

@@ -26,8 +26,8 @@ type (
  GetConfig() config.Blockchain
  GetDesignatedByRole(role noderoles.Role) (keys.PublicKeys, uint32, error)
  HeaderHeight() uint32
- SubscribeForBlocks(ch chan *block.Block)
- UnsubscribeFromBlocks(ch chan *block.Block)
+ SubscribeForBlocks(ch chan<- *block.Block)
+ UnsubscribeFromBlocks(ch chan<- *block.Block)
  }
 
  // Service represents a state root service.