Method GetAccountBalancesStream

engine/rpcserver.go:665–715

GetAccountBalancesStream streams an account balance for a specific exchange

(r *gctrpc.GetAccountBalancesRequest, stream gctrpc.GoCryptoTraderService_GetAccountBalancesStreamServer)

// GetAccountBalancesStream streams an account balance for a specific exchange
func (s *RPCServer) GetAccountBalancesStream(r *gctrpc.GetAccountBalancesRequest, stream gctrpc.GoCryptoTraderService_GetAccountBalancesStreamServer) error {
	assetType, err := asset.New(r.AssetType)
	if err != nil {
		return err
	}

	exch, err := s.GetExchangeByName(r.Exchange)
	if err != nil {
		return err
	}

	err = checkParams(r.Exchange, exch, assetType, currency.EMPTYPAIR)
	if err != nil {
		return err
	}

	pipe, err := exch.SubscribeAccountBalances()
	if err != nil {
		return err
	}

	defer func() {
		pipeErr := pipe.Release()
		if pipeErr != nil {
			log.Errorln(log.DispatchMgr, pipeErr)
		}
	}()
	init := make(chan struct{}, 1)
	init <- struct{}{}

	for {
		select {
		case <-stream.Context().Done():
			return stream.Context().Err()
		case _, ok := <-pipe.Channel():
			if !ok {
				return errDispatchSystem
			}
		case <-init:
		}

		subAccts, err := exch.GetCachedSubAccounts(stream.Context(), assetType)
		if err != nil {
			return err
		}

		if err := stream.Send(accountBalanceResp(r.Exchange, subAccts)); err != nil {
			return err
		}
	}
}

// GetConfig returns the bots config
func (s *RPCServer) GetConfig(_ context.Context, _ *gctrpc.GetConfigRequest) (*gctrpc.GetConfigResponse, error) {