diff --git a/go/src/mapping/channel_mapper.go b/go/src/mapping/channel_mapper.go index 0b0dc8b03..53c014456 100644 --- a/go/src/mapping/channel_mapper.go +++ b/go/src/mapping/channel_mapper.go @@ -11,10 +11,9 @@ func ChannelMapper( ) { for { lrec := <-inrecs - if lrec == nil { - outrecs <- nil + recordMapper.Map(lrec, outrecs) + if lrec == nil { // end of stream break } - recordMapper.Map(lrec, outrecs) } } diff --git a/go/src/mapping/mapper_cat.go b/go/src/mapping/mapper_cat.go new file mode 100644 index 000000000..faf8b6f5f --- /dev/null +++ b/go/src/mapping/mapper_cat.go @@ -0,0 +1,18 @@ +package mapping + +import ( + "containers" +) + +type MapperCat struct { + // stateless +} + +func NewMapperCat() *MapperCat { + return &MapperCat { + } +} + +func (this *MapperCat) Map(inrec *containers.Lrec, outrecs chan<- *containers.Lrec) { + outrecs <- inrec +} diff --git a/go/src/mapping/mapper_tac.go b/go/src/mapping/mapper_tac.go new file mode 100644 index 000000000..359e79aac --- /dev/null +++ b/go/src/mapping/mapper_tac.go @@ -0,0 +1,30 @@ +package mapping + +import ( + // System: + "container/list" + // Miller: + "containers" +) + +type MapperTac struct { + lrecs *list.List +} + +func NewMapperTac() *MapperTac { + return &MapperTac { + list.New(), + } +} + +func (this *MapperTac) Map(inrec *containers.Lrec, outrecs chan<- *containers.Lrec) { + if inrec != nil { + this.lrecs.PushFront(inrec) + } else { + // end of stream + for e := this.lrecs.Front(); e != nil; e = e.Next() { + outrecs <- e.Value.(*containers.Lrec) + } + outrecs <- nil + } +} diff --git a/go/src/stream/stream.go b/go/src/stream/stream.go index 761747dde..6ddbaf56e 100644 --- a/go/src/stream/stream.go +++ b/go/src/stream/stream.go @@ -26,7 +26,9 @@ func Stream(filenames []string) error { outrecs := make(chan *containers.Lrec, 1) donechan := make(chan bool, 1) - recordMapper := mapping.NewMapperFoo(); + //recordMapper := mapping.NewMapperFoo(); + //recordMapper := mapping.NewMapperCat(); + recordMapper := mapping.NewMapperTac(); go input.ChannelReader(reader, inrecs, echan) go mapping.ChannelMapper(inrecs, recordMapper, outrecs)