diff --git a/VERSION b/VERSION
index 04cc999..545fd57 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-1.15.6
+1.15.7
diff --git a/internal/bbsim/api/onus_handler.go b/internal/bbsim/api/onus_handler.go
index 72d34ec..58946c5 100644
--- a/internal/bbsim/api/onus_handler.go
+++ b/internal/bbsim/api/onus_handler.go
@@ -231,7 +231,6 @@
 func (s BBSimServer) ChangeIgmpState(ctx context.Context, req *bbsim.IgmpRequest) (*bbsim.Response, error) {
 
 	// NOTE this API will change the IGMP state for all UNIs on the requested ONU
-	// TODO a new API needs to be created to individually manage the UNIs
 
 	res := &bbsim.Response{}
 
@@ -312,18 +311,21 @@
 
 				switch req.SubActionVal {
 				case bbsim.SubActionTypes_JOIN:
+					service.AddGroupAddress(req.GroupAddress, ctag)
 					go func() {
 						_ = igmp.SendIGMPMembershipReportV2(service.UniPort.Onu.PonPortID, service.UniPort.Onu.ID, service.UniPort.Onu.Sn(),
 							service.UniPort.PortNo, service.UniPort.ID, service.GemPort, service.HwAddress, ctag,
 							service.UsPonCTagPriority, service.Stream, req.GroupAddress)
 					}()
 				case bbsim.SubActionTypes_LEAVE:
+					service.RemoveGroupAddress(req.GroupAddress)
 					go func() {
 						_ = igmp.SendIGMPLeaveGroupV2(service.UniPort.Onu.PonPortID, service.UniPort.Onu.ID, service.UniPort.Onu.Sn(),
 							service.UniPort.PortNo, service.UniPort.ID, service.GemPort, service.HwAddress, ctag,
 							service.UsPonCTagPriority, service.Stream, req.GroupAddress)
 					}()
 				case bbsim.SubActionTypes_JOINV3:
+					service.AddGroupAddress(req.GroupAddress, ctag)
 					go func() {
 						_ = igmp.SendIGMPMembershipReportV3(service.UniPort.Onu.PonPortID, service.UniPort.Onu.ID, service.UniPort.Onu.Sn(),
 							service.UniPort.PortNo, service.UniPort.ID, service.GemPort, service.HwAddress, ctag,
diff --git a/internal/bbsim/devices/olt.go b/internal/bbsim/devices/olt.go
index 587a2ed..96ee5f1 100644
--- a/internal/bbsim/devices/olt.go
+++ b/internal/bbsim/devices/olt.go
@@ -19,6 +19,7 @@
 import (
 	"context"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"net"
 	"strconv"
@@ -51,6 +52,10 @@
 })
 
 const (
+	multicastGemPortId = 4069
+)
+
+const (
 	//InternalState FSM states and transitions
 	OltInternalStateCreated     = "created"
 	OltInternalStateInitialized = "initialized"
@@ -1439,49 +1444,64 @@
 	pon, err := o.GetPonById(onuPkt.IntfId)
 	if err != nil {
 		oltLogger.WithFields(log.Fields{
-			"OnuId":  onuPkt.OnuId,
-			"IntfId": onuPkt.IntfId,
-			"err":    err,
+			"OnuId":     onuPkt.OnuId,
+			"IntfId":    onuPkt.IntfId,
+			"GemportId": onuPkt.GemportId,
+			"err":       err,
 		}).Error("Can't find PonPort")
 	}
-	onu, err := pon.GetOnuById(onuPkt.OnuId)
-	if err != nil {
-		oltLogger.WithFields(log.Fields{
-			"OnuId":  onuPkt.OnuId,
-			"IntfId": onuPkt.IntfId,
-			"err":    err,
-		}).Error("Can't find Onu")
-	}
 
-	oltLogger.WithFields(log.Fields{
-		"IntfId": onu.PonPortID,
-		"OnuId":  onu.ID,
-		"OnuSn":  onu.Sn(),
-		"Packet": hex.EncodeToString(onuPkt.Pkt),
-	}).Trace("Received OnuPacketOut")
+	onus := make([]*Onu, 1)
+	// If it's not addressed to multicast gem port
+	if onuPkt.GemportId != multicastGemPortId {
+		onus[0], err = pon.GetOnuById(onuPkt.OnuId)
+		if err != nil {
+			oltLogger.WithFields(log.Fields{
+				"OnuId":     onuPkt.OnuId,
+				"IntfId":    onuPkt.IntfId,
+				"GemportId": onuPkt.GemportId,
+				"err":       err,
+			}).Error("Can't find Onu")
+			return new(openolt.Empty), errors.New("cant-find-onu-by-id")
+		}
+		oltLogger.WithFields(log.Fields{
+			"IntfId":    onus[0].PonPortID,
+			"OnuId":     onus[0].ID,
+			"OnuSn":     onus[0].Sn(),
+			"GemportId": onuPkt.GemportId,
+			"Packet":    hex.EncodeToString(onuPkt.Pkt),
+		}).Trace("Received OnuPacketOut")
+	} else {
+		onus = pon.GetAllOnus()
+		oltLogger.WithFields(log.Fields{
+			"IntfId":    onuPkt.IntfId,
+			"GemportId": onuPkt.GemportId,
+			"Packet":    hex.EncodeToString(onuPkt.Pkt),
+		}).Trace("Received OnuPacketOut to multicast gem port")
+	}
 
 	rawpkt := gopacket.NewPacket(onuPkt.Pkt, layers.LayerTypeEthernet, gopacket.Default)
 
 	pktType, err := packetHandlers.GetPktType(rawpkt)
 	if err != nil {
 		onuLogger.WithFields(log.Fields{
-			"IntfId": onu.PonPortID,
-			"OnuId":  onu.ID,
-			"OnuSn":  onu.Sn(),
-			"Pkt":    hex.EncodeToString(rawpkt.Data()),
-		}).Error("Can't find pktType in packet, droppint it")
-		return new(openolt.Empty), nil
+			"IntfId":    onuPkt.IntfId,
+			"OnuId":     onuPkt.OnuId,
+			"GemportId": onuPkt.GemportId,
+			"Pkt":       hex.EncodeToString(rawpkt.Data()),
+		}).Debug("Can't find pktType in packet, dropping it")
+		return new(openolt.Empty), errors.New("malformed-packet")
 	}
 
 	pktMac, err := packetHandlers.GetDstMacAddressFromPacket(rawpkt)
 	if err != nil {
 		onuLogger.WithFields(log.Fields{
-			"IntfId": onu.PonPortID,
-			"OnuId":  onu.ID,
-			"OnuSn":  onu.Sn(),
-			"Pkt":    rawpkt.Data(),
-		}).Error("Can't find Dst MacAddress in packet, droppint it")
-		return new(openolt.Empty), nil
+			"IntfId":    onuPkt.IntfId,
+			"OnuId":     onuPkt.OnuId,
+			"GemportId": onuPkt.GemportId,
+			"Pkt":       rawpkt.Data(),
+		}).Debug("Can't find Dst MacAddress in packet, droppint it")
+		return new(openolt.Empty), errors.New("dst-mac-can-not-found-in-packet")
 	}
 
 	msg := types.Message{
@@ -1496,7 +1516,22 @@
 		},
 	}
 
-	onu.Channel <- msg
+	for _, onu := range onus {
+		if onu.InternalState.Current() == OnuStateEnabled {
+			oltLogger.WithFields(log.Fields{
+				"IntfId": onu.PonPortID,
+				"OnuId":  onu.ID,
+				"OnuSn":  onu.Sn(),
+			}).Trace("Sending to onuchannel")
+			onu.Channel <- msg
+		} else {
+			oltLogger.WithFields(log.Fields{
+				"IntfId": onu.PonPortID,
+				"OnuId":  onu.ID,
+				"OnuSn":  onu.Sn(),
+			}).Debug("can-not-send-onu-packet-out-to-onu")
+		}
+	}
 
 	return new(openolt.Empty), nil
 }
diff --git a/internal/bbsim/devices/onu.go b/internal/bbsim/devices/onu.go
index a12bd26..ebe08d6 100644
--- a/internal/bbsim/devices/onu.go
+++ b/internal/bbsim/devices/onu.go
@@ -508,26 +508,45 @@
 				msg, _ := message.Data.(bbsim.OnuPacketMessage)
 
 				onuLogger.WithFields(log.Fields{
-					"IntfId":  msg.IntfId,
-					"OnuId":   msg.OnuId,
-					"pktType": msg.Type,
+					"IntfId":    msg.IntfId,
+					"OnuId":     msg.OnuId,
+					"pktType":   msg.Type,
+					"OnuSn":     o.Sn(),
+					"gemportid": msg.GemPortId,
 				}).Trace("Received OnuPacketOut Message")
 
-				uni, err := o.findUniByPortNo(msg.PortNo)
-
-				if err != nil {
-					onuLogger.WithFields(log.Fields{
-						"IntfId":     msg.IntfId,
-						"OnuId":      msg.OnuId,
-						"pktType":    msg.Type,
-						"portNo":     msg.PortNo,
-						"MacAddress": msg.MacAddress,
-						"Pkt":        hex.EncodeToString(msg.Packet.Data()),
-						"OnuSn":      o.Sn(),
-					}).Error("Cannot find Uni associated with packet")
-					return
+				if msg.GemPortId == multicastGemPortId {
+					unis := o.findUniWithIgmpMembership()
+					if len(unis) == 0 {
+						onuLogger.WithFields(log.Fields{
+							"IntfId":     msg.IntfId,
+							"OnuId":      msg.OnuId,
+							"pktType":    msg.Type,
+							"portNo":     msg.PortNo,
+							"MacAddress": msg.MacAddress,
+							"Pkt":        hex.EncodeToString(msg.Packet.Data()),
+							"OnuSn":      o.Sn(),
+						}).Trace("No uni to forward msg coming to multicast gemport")
+					}
+					for _, uni := range unis {
+						uni.PacketCh <- msg
+					}
+				} else {
+					uni, err := o.findUniByPortNo(msg.PortNo)
+					if err != nil {
+						onuLogger.WithFields(log.Fields{
+							"IntfId":     msg.IntfId,
+							"OnuId":      msg.OnuId,
+							"pktType":    msg.Type,
+							"portNo":     msg.PortNo,
+							"MacAddress": msg.MacAddress,
+							"Pkt":        hex.EncodeToString(msg.Packet.Data()),
+							"OnuSn":      o.Sn(),
+						}).Error("Cannot find Uni associated with packet")
+						continue
+					}
+					uni.PacketCh <- msg
 				}
-				uni.PacketCh <- msg
 			// BBR specific messages
 			case bbsim.OnuPacketIn:
 				// NOTE we only receive BBR packets here.
@@ -1863,6 +1882,38 @@
 	return nil, fmt.Errorf("cannot-find-uni-with-port-no-%d", portNo)
 }
 
+// findUniWithIgmpMembership returns the list of UNIs which has sent any IGMP messages
+// and has any active membershipments currently
+func (onu *Onu) findUniWithIgmpMembership() []UniPort {
+	var uniPorts []UniPort
+	for _, u := range onu.UniPorts {
+		uni := u.(*UniPort)
+		if !uni.OperState.Is(UniStateUp) {
+			// if the UNI is disabled, ignore it
+			continue
+		}
+		for _, s := range uni.Services {
+			service := s.(*Service)
+			if service.NeedsIgmp {
+				if !service.InternalState.Is(ServiceStateInitialized) {
+					log.WithFields(log.Fields{
+						"OnuId":   onu.ID,
+						"UniId":   uni.ID,
+						"IntfId":  onu.PonPortID,
+						"OnuSn":   onu.Sn(),
+						"Service": service.Name,
+					}).Warn("service-not-initialized-skipping")
+					continue
+				}
+				if len(service.groupAddresses) > 0 {
+					uniPorts = append(uniPorts, *uni)
+				}
+			}
+		}
+	}
+	return uniPorts
+}
+
 func (o *Onu) SendOMCIAlarmNotificationMsg(raiseOMCIAlarm bool, alarmType string) {
 	switch alarmType {
 	case "ONU_ALARM_LOS":
diff --git a/internal/bbsim/devices/pon.go b/internal/bbsim/devices/pon.go
index 23021a1..0caaa42 100644
--- a/internal/bbsim/devices/pon.go
+++ b/internal/bbsim/devices/pon.go
@@ -67,6 +67,10 @@
 	allocatedAllocIdsLock sync.RWMutex
 }
 
+func (p *PonPort) GetAllOnus() []*Onu {
+	return p.Onus
+}
+
 // CreatePonPort creates pon port object
 func CreatePonPort(olt *OltDevice, id uint32, tech common.PonTechnology) *PonPort {
 	ponPort := PonPort{
diff --git a/internal/bbsim/devices/services.go b/internal/bbsim/devices/services.go
index 4ac1d39..788f05b 100644
--- a/internal/bbsim/devices/services.go
+++ b/internal/bbsim/devices/services.go
@@ -17,7 +17,6 @@
 package devices
 
 import (
-	"encoding/hex"
 	"fmt"
 	"net"
 	"time"
@@ -82,8 +81,12 @@
 	DsPonCTagPriority   uint8
 	DsPonSTagPriority   uint8
 
+	// gem port received in flow add requests
+	GemPort uint32
+	// Multicast addresses joined with the ctag
+	groupAddresses map[string]int
+
 	// state
-	GemPort       uint32
 	InternalState *fsm.FSM
 	EapolState    *fsm.FSM
 	DHCPState     *fsm.FSM
@@ -155,6 +158,9 @@
 
 				service.PacketCh = nil
 				service.Channel = nil
+
+				service.GemPort = 0
+				service.groupAddresses = nil
 			},
 		},
 	)
@@ -470,8 +476,7 @@
 
 			_ = dhcp.HandleNextPacket(s.UniPort.Onu.ID, s.UniPort.Onu.PonPortID, s.Name, s.UniPort.Onu.Sn(), s.UniPort.PortNo, tag, s.GemPort, s.UniPort.ID, s.HwAddress, s.DHCPState, msg.Packet, priority, s.Stream)
 		} else if msg.Type == packetHandlers.IGMP {
-			log.Warn(hex.EncodeToString(msg.Packet.Data()))
-			_ = igmp.HandleNextPacket(s.UniPort.Onu.PonPortID, s.UniPort.Onu.ID, s.UniPort.Onu.Sn(), s.UniPort.PortNo, s.UniPort.ID, s.GemPort, s.HwAddress, msg.Packet, s.CTag, s.UsPonCTagPriority, s.Stream)
+			_ = igmp.HandleNextPacket(s.UniPort.Onu.PonPortID, s.UniPort.Onu.ID, s.UniPort.Onu.Sn(), s.UniPort.PortNo, s.UniPort.ID, s.GemPort, s.HwAddress, msg.Packet, s.CTag, s.UsPonCTagPriority, s.groupAddresses, s.Stream)
 		}
 	}
 }
@@ -529,6 +534,7 @@
 				_ = s.DHCPState.Event("dhcp_failed")
 
 			}
+		// IGMP block is not in use since the event mechanism is disable for IGMP
 		case bbsimTypes.IGMPMembershipReportV2:
 			igmpInfo, _ := msg.Data.(bbsimTypes.IgmpMessage)
 			serviceLogger.WithFields(log.Fields{
@@ -552,6 +558,7 @@
 				"UniId":     s.UniPort.ID,
 				"Name":      s.Name,
 			}).Debug("Received IGMPLeaveGroupV2 message on ONU channel")
+			s.RemoveGroupAddress(igmpInfo.GroupAddress)
 			_ = igmp.SendIGMPLeaveGroupV2(s.UniPort.Onu.PonPortID, s.UniPort.Onu.ID, s.UniPort.Onu.Sn(), s.UniPort.PortNo, s.UniPort.ID, s.GemPort, s.HwAddress, s.CTag, s.UsPonCTagPriority, s.Stream, igmpInfo.GroupAddress)
 		case bbsimTypes.IGMPMembershipReportV3:
 			igmpInfo, _ := msg.Data.(bbsimTypes.IgmpMessage)
@@ -671,3 +678,17 @@
 		"Name":      s.Name,
 	}).Debugf("Changing Service.%s InternalState from %s to %s", stateMachine, src, dst)
 }
+
+func (s *Service) AddGroupAddress(addr string, ctag int) {
+	if s.groupAddresses == nil {
+		s.groupAddresses = make(map[string]int)
+	}
+	s.groupAddresses[addr] = ctag
+}
+
+func (s *Service) RemoveGroupAddress(addr string) {
+	if s.groupAddresses == nil {
+		return
+	}
+	delete(s.groupAddresses, addr)
+}
diff --git a/internal/bbsim/responders/igmp/igmp.go b/internal/bbsim/responders/igmp/igmp.go
index fe30cd5..1461a8c 100644
--- a/internal/bbsim/responders/igmp/igmp.go
+++ b/internal/bbsim/responders/igmp/igmp.go
@@ -196,7 +196,14 @@
 }
 
 func HandleNextPacket(ponPortId uint32, onuId uint32, serialNumber string, portNo uint32, uniId uint32,
-	gemPortId uint32, macAddress net.HardwareAddr, pkt gopacket.Packet, cTag int, pbit uint8, stream bbsim.Stream) error {
+	gemPortId uint32, macAddress net.HardwareAddr, pkt gopacket.Packet, cTag int, pbit uint8, groupAddresses map[string]int, stream bbsim.Stream) error {
+
+	dot1qLayer := pkt.Layer(layers.LayerTypeDot1Q)
+	if dot1qLayer == nil {
+		log.WithFields(log.Fields{"Pkt": hex.EncodeToString(pkt.Data())}).Warnf("packet-is-not-tagged")
+		return errors.New("packet-is-not-tagged")
+	}
+	dot1q := dot1qLayer.(*layers.Dot1Q)
 
 	igmpLayer := pkt.Layer(layers.LayerTypeIGMP)
 	if igmpLayer == nil {
@@ -210,14 +217,57 @@
 	}
 
 	log.WithFields(log.Fields{
-		"Pkt": pkt.Data(),
+		"Pkt": hex.EncodeToString(pkt.Data()),
 	}).Trace("IGMP packet")
 
-	igmp := igmpLayer.(*layers.IGMPv1or2)
+	var igmp *IGMP
+	igmpv1or2, ok := igmpLayer.(*layers.IGMPv1or2)
+	if ok {
+		igmp = &IGMP{
+			Type:            igmpv1or2.Type,
+			MaxResponseTime: igmpv1or2.MaxResponseTime,
+			Checksum:        igmpv1or2.Checksum,
+			GroupAddress:    igmpv1or2.GroupAddress,
+			Version:         igmpv1or2.Version,
+		}
+	} else {
+		igmpv3, ok := igmpLayer.(*layers.IGMP)
+		if ok {
+			igmp = &IGMP{
+				Type:                    igmpv3.Type,
+				MaxResponseTime:         igmpv3.MaxResponseTime,
+				Checksum:                igmpv3.Checksum,
+				GroupAddress:            igmpv3.GroupAddress,
+				SupressRouterProcessing: igmpv3.SupressRouterProcessing,
+				RobustnessValue:         igmpv3.RobustnessValue,
+				IntervalTime:            igmpv3.IntervalTime,
+				SourceAddresses:         igmpv3.SourceAddresses,
+				NumberOfGroupRecords:    igmpv3.NumberOfGroupRecords,
+				NumberOfSources:         igmpv3.NumberOfSources,
+				GroupRecords:            igmpv3.GroupRecords,
+				Version:                 igmpv3.Version,
+			}
+		} else {
+			log.Warnf("could-not-parse-igmp-packet")
+		}
+	}
 
 	if igmp.Type == layers.IGMPMembershipQuery {
-		_ = SendIGMPMembershipReportV2(ponPortId, onuId, serialNumber, portNo, uniId, gemPortId, macAddress,
-			cTag, pbit, stream, igmp.GroupAddress.String())
+		// Send response to queries only if joined to a group
+		for groupAddr, vlan := range groupAddresses {
+			if vlan == int(dot1q.VLANIdentifier) {
+				_ = SendIGMPMembershipReportV2(ponPortId, onuId, serialNumber, portNo, uniId, gemPortId, macAddress,
+					vlan, pbit, stream, groupAddr)
+			}
+		}
+		if groupAddresses == nil {
+			log.WithFields(log.Fields{
+				"OnuId":        onuId,
+				"SerialNumber": serialNumber,
+				"PortNo":       portNo,
+				"Pkt":          hex.EncodeToString(pkt.Data()),
+			}).Trace("no-active-channels")
+		}
 	}
 
 	return nil
@@ -225,21 +275,30 @@
 
 func createIGMPV3MembershipReportPacket(groupAddress string) *IGMP {
 
-	groupRecord1 := IGMPv3GroupRecord{
-		Type:             IGMPv3GroupRecordType(IGMPIsIn),
-		AuxDataLen:       0, // this should always be 0 as per IGMPv3 spec.
-		NumberOfSources:  3,
-		MulticastAddress: net.IPv4(224, 0, 0, 22),
-		SourceAddresses:  []net.IP{net.IPv4(15, 14, 20, 24), net.IPv4(15, 14, 20, 26), net.IPv4(15, 14, 20, 25)},
-		AuxData:          0, // NOT USED
-	}
+	// groupRecord1 := IGMPv3GroupRecord{
+	// 	Type:             IGMPv3GroupRecordType(IGMPIsIn),
+	// 	AuxDataLen:       0, // this should always be 0 as per IGMPv3 spec.
+	// 	NumberOfSources:  3,
+	// 	MulticastAddress: net.IPv4(224, 0, 0, 22),
+	// 	SourceAddresses:  []net.IP{net.IPv4(15, 14, 20, 24), net.IPv4(15, 14, 20, 26), net.IPv4(15, 14, 20, 25)},
+	// 	AuxData:          0, // NOT USED
+	// }
 
-	groupRecord2 := IGMPv3GroupRecord{
-		Type:             IGMPv3GroupRecordType(IGMPIsIn),
+	// Keeping this as an example
+	// groupRecord1 := IGMPv3GroupRecord{
+	// 	Type:             IGMPv3GroupRecordType(IGMPIsIn),
+	// 	AuxDataLen:       0, // this should always be 0 as per IGMPv3 spec.
+	// 	NumberOfSources:  3,
+	// 	MulticastAddress: net.IPv4(224, 0, 0, 22),
+	// 	SourceAddresses:  []net.IP{net.IPv4(15, 14, 20, 24), net.IPv4(15, 14, 20, 26), net.IPv4(15, 14, 20, 25)},
+	// 	AuxData:          0, // NOT USED
+	// }
+
+	groupRecord := layers.IGMPv3GroupRecord{
+		Type:             layers.IGMPv3GroupRecordType(IGMPIsEx),
 		AuxDataLen:       0, // this should always be 0 as per IGMPv3 spec.
-		NumberOfSources:  2,
-		MulticastAddress: net.IPv4(224, 0, 0, 25),
-		SourceAddresses:  []net.IP{net.IPv4(15, 14, 20, 30), net.IPv4(15, 14, 20, 31)},
+		NumberOfSources:  0,
+		MulticastAddress: net.ParseIP(groupAddress),
 		AuxData:          0, // NOT USED
 	}
 
@@ -251,10 +310,9 @@
 		SupressRouterProcessing: false,
 		RobustnessValue:         0,
 		IntervalTime:            time.Duration(1),
-		SourceAddresses:         []net.IP{net.IPv4(224, 0, 0, 24)},
-		NumberOfGroupRecords:    2,
-		NumberOfSources:         1,
-		GroupRecords:            []IGMPv3GroupRecord{groupRecord1, groupRecord2},
+		NumberOfGroupRecords:    1,
+		NumberOfSources:         0,
+		GroupRecords:            []layers.IGMPv3GroupRecord{groupRecord},
 		Version:                 3,
 	}
 
@@ -339,7 +397,7 @@
 	SourceAddresses         []net.IP
 	NumberOfGroupRecords    uint16
 	NumberOfSources         uint16
-	GroupRecords            []IGMPv3GroupRecord
+	GroupRecords            []layers.IGMPv3GroupRecord
 	Version                 uint8 // IGMP protocol version
 }
 
diff --git a/internal/bbsim/responders/igmp/igmp_test.go b/internal/bbsim/responders/igmp/igmp_test.go
index d6442a3..7319db8 100644
--- a/internal/bbsim/responders/igmp/igmp_test.go
+++ b/internal/bbsim/responders/igmp/igmp_test.go
@@ -65,7 +65,7 @@
 
 	fmt.Println(packet.Layers())
 
-	err := HandleNextPacket(0, 0, "FOO", 1, 1024, 0, mac, packet, 55, 5, stream)
+	err := HandleNextPacket(0, 0, "FOO", 1, 1024, 0, mac, packet, 55, 5, nil, stream)
 	assert.Nil(t, err)
 
 	assert.Equal(t, 1, stream.CallCount)