src/qpid/amqp_0_10/Body.h, Header.h: placeholders.

src/qpid/amqp_0_10/Unit.h,.cpp
 - Decode "units" - command/control/header segments or body frames.
   Equivalent to preview AMQFrame class.


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@645733 13f79535-47bb-0310-9956-ffa450edef68

14 files changed, 455 insertions, 429 deletions

diff --git a/cpp/rubygen/0-10/specification.rb b/cpp/rubygen/0-10/specification.rb
index 83a7da07a2..d8c5378538 100755
--- a/cpp/rubygen/0-10/specification.rb
+++ b/cpp/rubygen/0-10/specification.rb
@@ -65,6 +65,7 @@ class Specification < CppGen
       yield if block
     }
     genl "std::ostream& operator << (std::ostream&, const #{x.classname}&);"
+    genl "bool operator==(const #{x.classname}&, const #{x.classname}&);"
   end
 
   # FIXME aconway 2008-03-10: packing, coding
diff --git a/cpp/src/Makefile.am b/cpp/src/Makefile.am
index 4aed05a1ef..991d9bc4c5 100644
--- a/cpp/src/Makefile.am
+++ b/cpp/src/Makefile.am
@@ -107,16 +107,19 @@ libqpidcommon_la_SOURCES = \
   qpid/amqp_0_10/built_in_types.h \
   qpid/amqp_0_10/complex_types.h \
   qpid/amqp_0_10/complex_types.cpp \
+  qpid/amqp_0_10/Body.h \
+  qpid/amqp_0_10/Header.h \
+  qpid/amqp_0_10/FrameHeader.h \
+  qpid/amqp_0_10/FrameHeader.cpp \
   qpid/amqp_0_10/Holder.h \
   qpid/amqp_0_10/Codec.h \
   qpid/amqp_0_10/Packer.h \
   qpid/amqp_0_10/Decimal.h \
-  qpid/amqp_0_10/Frame.h \
-  qpid/amqp_0_10/Segment.h \
-  qpid/amqp_0_10/Segment.cpp \
   qpid/amqp_0_10/SerializableString.h \
   qpid/amqp_0_10/Map.h \
   qpid/amqp_0_10/Map.cpp \
+  qpid/amqp_0_10/Unit.h \
+  qpid/amqp_0_10/Unit.cpp \
   qpid/amqp_0_10/UnknownType.h \
   qpid/amqp_0_10/UnknownType.cpp \
   qpid/Serializer.h \
diff --git a/cpp/src/qpid/amqp_0_10/Body.h b/cpp/src/qpid/amqp_0_10/Body.h
new file mode 100644
index 0000000000..a2192503b8
--- /dev/null
+++ b/cpp/src/qpid/amqp_0_10/Body.h
@@ -0,0 +1,55 @@
+#ifndef QPID_AMQP_0_10_BODY_H
+#define QPID_AMQP_0_10_BODY_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include <string>
+#include <ostream>
+
+namespace qpid {
+namespace amqp_0_10 {
+
+/** Holds data from a body frame. */
+class Body {
+  public:
+    Body() {}
+    Body(size_t size_) : str('\0', size_) {}
+    Body(const char* data_, size_t size_) : str(data_, size_) {}
+
+    size_t size() const { return str.size(); };
+    const char* data() const { return str.data(); }
+    char* data() { return const_cast<char*>(str.data()); }
+
+    template <class S> void serialize(S& s) { s.raw(data(), size()); }
+    
+  private:
+    std::string str;
+
+  friend std::ostream& operator<<(std::ostream&, const Body&);
+};
+
+inline std::ostream& operator<<(std::ostream& o, const Body& b) {
+    return o << b.str.substr(16) << "... (" << b.size() << ")";
+}
+
+}} // namespace qpid::amqp_0_10
+
+#endif  /*!QPID_AMQP_0_10_BODY_H*/
diff --git a/cpp/src/qpid/amqp_0_10/Frame.h b/cpp/src/qpid/amqp_0_10/Frame.h
deleted file mode 100644
index f3a99e8004..0000000000
--- a/cpp/src/qpid/amqp_0_10/Frame.h
+++ /dev/null
@@ -1,172 +0,0 @@
-#ifndef QPID_AMQP_0_10_FRAME_H
-#define QPID_AMQP_0_10_FRAME_H
-
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include "qpid/amqp_0_10/built_in_types.h"
-#include <boost/shared_array.hpp>
-#include <string.h>
-
-namespace qpid {
-namespace amqp_0_10 {
-
-class FrameHeader {
-  public:
-    enum Flags { FIRST_SEGMENT=8, LAST_SEGMENT=4, FIRST_FRAME=2, LAST_FRAME=1 };
-    enum ShortFlags { FS=8, LS=4, FF=2, LF=1 };
-
-
-    FrameHeader(uint8_t flags_=0, SegmentType type_=SegmentType(), uint8_t track_=0, uint16_t channel_=0)
-        : flags(flags_), type(type_), size_(), track(track_), channel(channel_)
-    {}
-
-    uint8_t getFlags() const { return flags; }
-    SegmentType getType() const { return type; }
-    uint16_t getSize() const { return size_; }
-    uint8_t getTrack() const { return track; }
-    uint16_t getChannel() const { return channel; }
-
-    void setFlags(uint8_t flags_) { flags=flags_; }
-    void setType(SegmentType type_)  { type=type_; }
-    void setTrack(uint8_t track_) { track=track_; }
-    void setChannel(uint8_t channel_) { channel=channel_; }
-
-    Flags testFlags(uint8_t f) const { return Flags(flags & f); }
-    void raiseFlags(uint8_t f) { flags |= f; }
-    void clearFlags(uint8_t f) { flags &= ~f; }
-
-    uint16_t size() const { return size_; }
-    bool empty() const { return size()==0; }
-
-    bool operator==(const FrameHeader& x)
-    { return memcmp(this, &x, sizeof(this)); }
-
-    template <class S> void serialize(S& s);
-
-  protected:
-    uint8_t flags;
-    SegmentType type;
-    uint16_t size_;
-    uint8_t track;
-    uint16_t channel;
-
-  private:
-};
-
-class Frame : public FrameHeader {
-  public:
-    Frame(uint8_t flags_=0, SegmentType type_=SegmentType(), uint8_t track_=0,
-          uint16_t channel_=0)
-        : FrameHeader(flags_, type_, track_, channel_), ref() {}
-
-    Frame(const FrameHeader& header) : FrameHeader(header), ref() {}
-
-    Frame(const FrameHeader& header, size_t s)
-        : FrameHeader(header), ref() { resize(s); }
-
-    Frame(const FrameHeader& header, const char* a, const char* b,
-          bool copyFlag=true) : FrameHeader(header), ref() {
-        if (copyFlag) copy(a,b); else refer(a,b);
-    }
-    
-    Frame(const FrameHeader& header, const char* p, std::size_t s,
-          bool copyFlag=true) : FrameHeader(header), ref() {
-        if (copyFlag) copy(p,p+s); else refer(p,p+s);
-    }
-    
-
-    /** Allocate a buffer of at least size bytes */
-    void resize(uint16_t size);
-
-    /** Make the frame refer to byte range [begin,end)
-     * Range is NOT COPIED, it must be valid for lifetime of Frame.
-     */
-    void refer(const char* begin, const char* end);
-
-    /** Allocate a buffer and copy range begin/end */
-    void copy(const char* begin, const char* end);
-
-    char* begin() { assert(!ref); return data.get(); }
-    const char* begin() const { return ref ? ref : data.get(); }
-    char* end() { return begin() + size(); }
-    const char* end() const { return begin() + size(); }
-
-    void clear() { data.reset();  ref = 0; size_= 0; }
-    
-    template <class S> void serialize(S& s);
-    template <class S> void encode(S& s) const;
-    template <class S> void decode(S& s);
-
-  private:
-    boost::shared_array<char> data;
-    const char* ref;
-};
-
-inline void Frame::refer(const char* a, const char* b) {
-    data.reset();
-    ref = a;
-    size_ = b-a;
-}
-
-inline void Frame::copy(const char* a, const char* b) {
-    resize(b-a);
-    std::copy(a, b, begin());
-}
-
-inline void Frame::resize(uint16_t s) {
-    if (s > size() || ref) {
-        ref = 0;
-        data.reset(new char[s]);
-    }
-    size_=s;
-}
-
-template <class S> void FrameHeader::serialize(S& s) {
-    uint8_t pad8=0;
-    uint32_t pad32=0;
-    s(flags)(type)(size_)(pad8)(track)(channel)(pad32);
-}
-
-template <class S> void Frame::serialize(S& s) { s.split(*this); }
-
-template <class S> void Frame::encode(S& s) const {
-    s(static_cast<const FrameHeader&>(*this));
-    s.raw(begin(), size());
-}
-
-template <class S> void Frame::decode(S& s) {
-    try {
-        uint16_t oldSize = size_;
-        s(static_cast<FrameHeader&>(*this));
-        std::swap(oldSize, size_);
-        resize(oldSize);
-        s.raw(begin(), size());
-    } catch (...) {
-        clear();
-        throw;
-    }
-}
-
-}} // namespace qpid::amqp_0_10
-
-#endif  /*!QPID_AMQP_0_10_FRAME_H*/
-
diff --git a/cpp/src/qpid/amqp_0_10/FrameHeader.cpp b/cpp/src/qpid/amqp_0_10/FrameHeader.cpp
new file mode 100644
index 0000000000..f1a59b9e27
--- /dev/null
+++ b/cpp/src/qpid/amqp_0_10/FrameHeader.cpp
@@ -0,0 +1,50 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "FrameHeader.h"
+#include <ios>
+#include <iomanip>
+#include <ostream>
+
+using namespace std;
+
+namespace qpid {
+namespace amqp_0_10 {
+
+bool FrameHeader::operator==(const FrameHeader& x) const {
+    return flags == x.flags &&
+        type == x.type &&
+        size == x.size &&
+        track == x.track &&
+        channel == x.channel;
+}
+
+std::ostream& operator<<(std::ostream& o, const FrameHeader& f) {
+    std::ios::fmtflags saveFlags = o.flags();
+    return o << "Frame["
+             << "flags=" << std::hex << std::showbase << int(f.getFlags()) << std::setiosflags(saveFlags)
+             << " type=" << f.getType()
+             << " size=" << f.getSize()
+             << " track=" << int(f.getTrack())
+             << " channel=" << f.getChannel()
+             << "]";
+}
+
+}} // namespace qpid::amqp_0_10
diff --git a/cpp/src/qpid/amqp_0_10/FrameHeader.h b/cpp/src/qpid/amqp_0_10/FrameHeader.h
new file mode 100644
index 0000000000..b2f0619f9b
--- /dev/null
+++ b/cpp/src/qpid/amqp_0_10/FrameHeader.h
@@ -0,0 +1,90 @@
+#ifndef QPID_AMQP_0_10_FRAMEHEADER_H
+#define QPID_AMQP_0_10_FRAMEHEADER_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/amqp_0_10/built_in_types.h"
+#include <boost/shared_array.hpp>
+#include <string.h>
+#include <assert.h>
+#include <iosfwd>
+
+namespace qpid {
+namespace amqp_0_10 {
+
+enum FrameFlags { FIRST_SEGMENT=8, LAST_SEGMENT=4, FIRST_FRAME=2, LAST_FRAME=1 };
+
+class FrameHeader {
+  public:
+    static const size_t SIZE=12;
+    static uint8_t trackFor(SegmentType type) { return type == 0 ? 0 : 1; }
+    
+    FrameHeader(uint8_t flags_=0, SegmentType type_=SegmentType(), uint16_t size_=0, uint8_t track_=0, uint16_t channel_=0)
+        : flags(flags_), type(type_), size(size_), track(track_), channel(channel_)
+    {}
+
+    uint8_t getFlags() const { return flags; }
+    SegmentType getType() const { return type; }
+    /** @return size total size of of frame, including frame header. */
+    uint16_t getSize() const { return size; }
+    /** @return size of frame data, excluding frame header. */
+    uint16_t getDataSize() const { return size - SIZE; }
+    uint8_t getTrack() const { return track; }
+    uint16_t getChannel() const { return channel; }
+
+    void setFlags(uint8_t flags_) { flags=flags_; }
+    /** Also sets the track. There is no setTrack() */
+    void setType(SegmentType type_)  { type=type_; track=trackFor(type); }
+    /** @param size total size of of frame, including frame header. */
+    void setSize(uint16_t size_) { size = size_; }
+    /** @param size size of frame data, excluding frame header. */
+    void setDataSize(uint16_t size_) { size = size_+SIZE; }
+    void setChannel(uint8_t channel_) { channel=channel_; }
+
+    bool allFlags(uint8_t f) const { return (flags & f) == f; }
+    bool anyFlags(uint8_t f) const { return (flags & f); }
+
+    void raiseFlags(uint8_t f) { flags |= f; }
+    void clearFlags(uint8_t f) { flags &= ~f; }
+
+    bool isComplete() const { return allFlags(FIRST_FRAME | LAST_FRAME); }
+
+    bool operator==(const FrameHeader&) const;
+
+    template <class S> void serialize(S& s) {
+        uint8_t pad8=0; uint32_t pad32=0;
+        s(flags)(type)(size)(pad8)(track)(channel)(pad32);
+    }
+
+  private:
+    uint8_t flags;
+    SegmentType type;
+    uint16_t size;
+    uint8_t track;
+    uint16_t channel;
+};
+
+std::ostream& operator<<(std::ostream&, const FrameHeader&);
+
+}} // namespace qpid::amqp_0_10
+
+#endif  /*!QPID_AMQP_0_10_FRAMEHEADER_H*/
diff --git a/cpp/src/qpid/amqp_0_10/Header.h b/cpp/src/qpid/amqp_0_10/Header.h
new file mode 100644
index 0000000000..44edcb9f3d
--- /dev/null
+++ b/cpp/src/qpid/amqp_0_10/Header.h
@@ -0,0 +1,42 @@
+#ifndef QPID_AMQP_0_10_HEADER_H
+#define QPID_AMQP_0_10_HEADER_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include <ostream>
+
+namespace qpid {
+namespace amqp_0_10 {
+
+// FIXME aconway 2008-03-27: TODO
+class Header
+{
+  public:
+    template <class S> void serialize(S&) {}
+  private:
+};
+
+// FIXME aconway 2008-03-28: TODO
+inline std::ostream& operator<<(std::ostream& o, const Header&) { return o; }
+
+}} // namespace qpid::amqp_0_10
+
+#endif  /*!QPID_AMQP_0_10_HEADER_H*/
diff --git a/cpp/src/qpid/amqp_0_10/Segment.cpp b/cpp/src/qpid/amqp_0_10/Segment.cpp
deleted file mode 100644
index 8eebeb276d..0000000000
--- a/cpp/src/qpid/amqp_0_10/Segment.cpp
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-#include "Segment.h"
-#include "qpid/Exception.h"
-#include <boost/ref.hpp>
-#include <numeric>
-
-namespace qpid {
-namespace amqp_0_10 {
-
-Segment::Segment() : missing() {}
-
-bool Segment::isComplete() const {
-    return missing || (!frames.empty() && (frames.back().testFlags(Frame::LAST_FRAME)));
-}
-
-Segment::const_iterator Segment::begin() const {
-    return Segment::const_iterator(frames, frames.begin(), frames.begin()->begin());
-}
-
-Segment::const_iterator Segment::end() const {
-    return const_iterator(frames, frames.end(), frames.end()->end());
-}
-
-namespace {
-Frame::Flags segFlags(const Frame& f) {
-    return f.testFlags(Frame::FIRST_SEGMENT | Frame::LAST_SEGMENT);
-}
-} // namespace
-
-void Segment::add(const Frame& frame) {
-    // FIXME aconway 2008-03-07:  ex types & messages.
-    if (isComplete()) throw Exception("cannot add frame to complete segment");
-    if (!frames.empty()) {
-        if (frame.testFlags(Frame::FIRST_FRAME) ||
-            frame.getType() != frames.front().getType() ||
-            segFlags(frames.front()) != segFlags(frame))
-            throw Exception("invalid frame");
-    }
-    frames.push_back(frame);
-}
-
-bool Segment::isFirst() const {
-    return !frames.empty() && frames.front().testFlags(Frame::FIRST_SEGMENT);
-}
-
-bool Segment::isLast() const {
-    return !frames.empty() && frames.front().testFlags(Frame::LAST_SEGMENT);
-}
-
-namespace {
-size_t accumulate_size(size_t total, const Frame& f) {
-    return total+f.size();
-}
-}
-
-size_t Segment::size() const {
-    return std::accumulate(frames.begin(), frames.end(), 0, &accumulate_size);
-}
-
-}} // namespace qpid::amqp_0_10
-
-
diff --git a/cpp/src/qpid/amqp_0_10/Segment.h b/cpp/src/qpid/amqp_0_10/Segment.h
deleted file mode 100644
index 650573802b..0000000000
--- a/cpp/src/qpid/amqp_0_10/Segment.h
+++ /dev/null
@@ -1,128 +0,0 @@
-#ifndef QPID_AMQP_0_10_SEGMENT_H
-#define QPID_AMQP_0_10_SEGMENT_H
-
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- *
- */
-
-#include "Frame.h"
-#include "qpid/InlineVector.h"
-#include "qpid/amqp_0_10/built_in_types.h"
-#include <boost/iterator/iterator_facade.hpp>
-#include <boost/range/iterator_range.hpp>
-#include <vector>
-
-namespace qpid {
-namespace amqp_0_10 {
-
-/**
- * Represents a (possibly incomplete) header or body segment.
- */
-class Segment {
-  public:
-    // TODO aconway 2008-03-07: optimize - use ISlist instead of vector.
-    typedef std::vector<Frame> Frames; 
-    struct const_iterator;
-
-    Segment();
-    
-    bool isComplete() const;
-    bool isFirst() const;
-    bool isLast() const;
-
-    const Frames& getFrames() const { return frames; }
-
-    void add(const Frame& frame);
-
-    /** Mark an empty segment as complete.
-     *@pre frames.empty().
-     */
-    void setMissing() { assert(frames.empty()); missing = true; }
-
-    size_t size() const; ///< Content size.
-
-    /** Iterate over content as a continuous range of bytes. */
-    const_iterator begin() const;
-    const_iterator end() const;
-
-    // TODO aconway 2008-03-07: Provide output iterator
-    // that extends the segment as basis for ostream (XML exchange)
-  private:
-    bool missing;
-    Frames frames;
-};
-
-class Segment::const_iterator : public boost::iterator_facade<
-    Segment::const_iterator, const char, boost::forward_traversal_tag>
-{
-  public:
-    const_iterator() : frames(), p() {}
-
-  private:
-
-    void invariant() const {
-        assert(frames);
-        assert(frames->begin() <= i);
-        assert(i <= frames->end());
-        assert(i->begin() <= p);
-        assert(p <= i->end());
-    }
-    void valid() const {
-        invariant();
-        assert(p < i->end());
-    }
-
-    const_iterator(const Frames& f, Frames::const_iterator pos, const char* ptr)
-        : frames(&f), i(pos), p(ptr) { skip_empty(); }
-
-    const char& dereference() const { valid(); return *p; }
-    bool equal(const const_iterator& x) const { return p == x.p; }
-    void increment() { valid(); ++p; skip_empty(); }
-    void advance(ptrdiff_t n) {
-        ptrdiff_t r = i->end() - p;
-        while (n > r) {
-            assert(i != frames->end());
-            n -= r;
-            p = (++i)->begin();
-            r = i->size();
-        }
-        p += n;
-        skip_empty();
-    }
-    
-    void skip_empty() {
-        invariant();
-        while (p == i->end() && i != frames->end())
-            p = (++i)->begin();
-        invariant();
-    }
-
-    const Frames* frames;
-    Frames::const_iterator i;
-    const char* p;
-
-  friend class Segment;
-  friend class boost::iterator_core_access;
-};
-
-
-}} // namespace qpid::amqp_0_10
-
-#endif  /*!QPID_AMQP_0_10_SEGMENT_H*/
diff --git a/cpp/src/qpid/amqp_0_10/Unit.cpp b/cpp/src/qpid/amqp_0_10/Unit.cpp
new file mode 100644
index 0000000000..6b0a2e542d
--- /dev/null
+++ b/cpp/src/qpid/amqp_0_10/Unit.cpp
@@ -0,0 +1,56 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "Unit.h"
+#include "Codec.h"
+
+namespace qpid {
+namespace amqp_0_10 {
+
+void Unit::setVariant() {
+    switch (header.getType()) {
+      case CONTROL: variant=ControlHolder(); break;
+      case COMMAND: variant=CommandHolder();
+      case HEADER: variant=Header();
+      case BODY: variant=Body(header.getDataSize());
+    }
+}
+
+struct GetTypeVisitor : public boost::static_visitor<SegmentType> {
+    SegmentType operator()(const CommandHolder& ) const { return COMMAND; }
+    SegmentType operator()(const ControlHolder& ) const { return CONTROL; }
+    SegmentType operator()(const Header& ) const { return HEADER; }
+    SegmentType operator()(const Body&) const { return BODY; }
+};
+
+void Unit::setHeader(uint8_t flags) {
+    header.setFlags(flags);
+    GetTypeVisitor getter;
+    header.setType(variant.apply_visitor(getter));
+    header.setDataSize(Codec::size(*this));
+    // track automatically set from type.
+    // no channel specified at this point.
+}
+
+std::ostream& operator<<(std::ostream& o, const Unit& u) {
+    return o << u.getHeader() << " " << u.getVariant();
+}
+
+}} // namespace qpid::amqp_0_10
diff --git a/cpp/src/qpid/amqp_0_10/Unit.h b/cpp/src/qpid/amqp_0_10/Unit.h
new file mode 100644
index 0000000000..dad34fef01
--- /dev/null
+++ b/cpp/src/qpid/amqp_0_10/Unit.h
@@ -0,0 +1,76 @@
+#ifndef QPID_AMQP_0_10_UNIT_H
+#define QPID_AMQP_0_10_UNIT_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/amqp_0_10/ControlHolder.h"
+#include "qpid/amqp_0_10/CommandHolder.h"
+#include "qpid/amqp_0_10/Header.h"
+#include "qpid/amqp_0_10/Body.h"
+#include "qpid/amqp_0_10/FrameHeader.h"
+
+#include <boost/variant.hpp>
+#include <ostream>
+
+namespace qpid {
+namespace amqp_0_10 {
+
+
+/**
+ * A Unit contains a frame header and associated value.
+ * For all types except BODY the frame header is for a complete segment.
+ */
+class Unit {
+  public:
+    typedef boost::variant<ControlHolder, CommandHolder, Header, Body> Variant;
+
+    Unit(const FrameHeader& h=FrameHeader()) : header(h) { setVariant(); }
+
+    template <class T>
+    Unit(const T& t, uint8_t flags) : variant(t) { setHeader(flags); }
+
+    const FrameHeader& getHeader() const { return header; }
+
+    template<class T> const T* get() const { return boost::get<T>(&variant); }
+    template<class T> T* get() { return boost::get<T>(&variant); }
+    template<class T> Unit& operator==(const T& t) { variant=t; return *this; }
+    
+    template <class S> void serialize(S& s) { variant.apply_visitor(s); s.split(*this); }
+    template <class S> void encode(S&) const {} 
+    template <class S> void decode(S&) { setHeader(header.getFlags()); }
+
+    const Variant& getVariant() const { return variant; }
+    Variant& getVariant() { return variant; }
+    
+  private:
+    void setHeader(uint8_t flags);
+    void setVariant();
+    
+    Variant variant;
+    FrameHeader header;
+};
+
+std::ostream& operator<<(std::ostream& o, const Unit& u);
+    
+}} // namespace qpid::amqp_0_10
+
+#endif  /*!QPID_AMQP_0_10_UNIT_H*/
diff --git a/cpp/src/qpid/amqp_0_10/UnitHandler.h b/cpp/src/qpid/amqp_0_10/UnitHandler.h
new file mode 100644
index 0000000000..93a8ce573a
--- /dev/null
+++ b/cpp/src/qpid/amqp_0_10/UnitHandler.h
@@ -0,0 +1,35 @@
+#ifndef QPID_AMQP_0_10_UNITHANDLER_H
+#define QPID_AMQP_0_10_UNITHANDLER_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/framing/Handler.h"
+
+namespace qpid {
+namespace amqp_0_10 {
+
+class Unit;
+typedef framing::Handler<const Unit&> UnitHandler;
+
+}} // namespace qpid::amqp_0_10
+
+#endif  /*!QPID_AMQP_0_10_UNITHANDLER_H*/
diff --git a/cpp/src/tests/amqp_0_10/serialize.cpp b/cpp/src/tests/amqp_0_10/serialize.cpp
index fa00e79ff3..16559b3516 100644
--- a/cpp/src/tests/amqp_0_10/serialize.cpp
+++ b/cpp/src/tests/amqp_0_10/serialize.cpp
@@ -25,8 +25,9 @@
 #include "qpid/amqp_0_10/Codec.h"
 #include "qpid/amqp_0_10/specification.h"
 #include "qpid/amqp_0_10/ControlHolder.h"
-#include "qpid/amqp_0_10/Frame.h"
+#include "qpid/amqp_0_10/FrameHeader.h"
 #include "qpid/amqp_0_10/Map.h"
+#include "qpid/amqp_0_10/Unit.h"
 
 #include <boost/test/test_case_template.hpp>
 #include <boost/type_traits/is_arithmetic.hpp>
@@ -204,48 +205,6 @@ BOOST_AUTO_TEST_CASE(testControlEncodeDecode) {
     BOOST_CHECK_EQUAL(tune.heartbeatMax, 4u);
 }
 
-BOOST_AUTO_TEST_CASE(testFrameEncodeDecode) {
-    static const int overhead=12;
-    string data;
-    Frame r, c;
-    char d1[]="abcdefg";
-    r.refer(d1, d1+sizeof(d1));
-    r.setFlags(Frame::FIRST_FRAME);
-    r.setType(CONTROL);
-    r.setChannel(32);
-    r.setTrack(1);
-    char d2[]="01234567";
-    c.copy(d2, d2+sizeof(d2));
-
-    BOOST_CHECK_EQUAL(overhead+sizeof(d1), Codec::size(r));
-    BOOST_CHECK_EQUAL(overhead+sizeof(d2), Codec::size(c));
-    Codec::encode(std::back_inserter(data))(r)(c);
-    BOOST_CHECK_EQUAL(data.size(), Codec::size(r)+Codec::size(c));
-
-    FrameHeader fh;
-    std::string::iterator i = Codec::decode(data.begin())(fh).pos();
-    size_t s = fh.size();
-    BOOST_CHECK_EQUAL(s, sizeof(d1));
-    BOOST_CHECK_EQUAL(std::string(i, i+s), std::string(d1, d1+s));
-
-                      
-    Frame f1, f2;
-    Codec::decode(data.begin())(f1)(f2);
-    BOOST_CHECK_EQUAL(f1.size(), sizeof(d1));
-    BOOST_CHECK_EQUAL(std::string(f1.begin(), f1.size()),
-                      std::string(d1, sizeof(d1)));
-    BOOST_CHECK_EQUAL(f1.size(), r.size());
-    BOOST_CHECK_EQUAL(f1.getFlags(), Frame::FIRST_FRAME);
-    BOOST_CHECK_EQUAL(f1.getType(), CONTROL);
-    BOOST_CHECK_EQUAL(f1.getChannel(), 32);
-    BOOST_CHECK_EQUAL(f1.getTrack(), 1);
-
-    BOOST_CHECK_EQUAL(f2.size(), c.size());
-    BOOST_CHECK_EQUAL(std::string(f2.begin(), f2.end()),
-                      std::string(d2, d2+sizeof(d2)));
-    
-}
-
 struct DummyPacked {
     static const uint8_t PACK=1;
     boost::optional<char> i, j;
@@ -291,4 +250,37 @@ BOOST_AUTO_TEST_CASE(testPacked) {
     BOOST_CHECK_EQUAL(dummy.k, 'y');
 }
 
+BOOST_AUTO_TEST_CASE(testUnit) {
+    string data;
+    Control::Holder h(in_place<connection::Tune>(1,2,3,4));
+    Codec::encode(std::back_inserter(data))(h);
+
+    Unit unit(FrameHeader(FIRST_FRAME|LAST_FRAME, CONTROL));
+    Codec::decode(data.begin())(unit);
+
+    BOOST_REQUIRE(unit.get<ControlHolder>());
+
+    string data2;
+    Codec::encode(back_inserter(data2))(unit);
+    
+    BOOST_CHECK_EQUAL(data, data2);
+}
+
+// FIXME aconway 2008-04-07: TODO
+// BOOST_AUTO_TEST_CASE(testAllSegmentTypes) {
+//     string data;
+//     int n = allSegmentTypes(Codec::encode(data));
+
+//     string data2;
+//     Codec::Decoder<string::iterator> decode(data.begin(), data.size());
+//     while (decode.pos() != data.end()) {
+//         Unit unit;
+//         decode(unit);
+//         Codec::encode(back_insert(data));
+//         --n;
+//     }
+//     BOOST_CHECK_EQUAL(n, 0);
+//     BOOST_CHECK_EQUAL(data, data2);
+// }
+
 QPID_AUTO_TEST_SUITE_END()
diff --git a/cpp/src/tests/python_tests b/cpp/src/tests/python_tests
index 7e6ab8a312..a3478eac1c 100755
--- a/cpp/src/tests/python_tests
+++ b/cpp/src/tests/python_tests
@@ -2,10 +2,17 @@
 # Run the python tests.
 QPID_PORT=${QPID_PORT:-5672}
 PYTHON_TESTS=${PYTHON_TESTS:-$*}
+
+run() {
+    SPEC=$1
+    FAILING=$2
+    ./run-tests --skip-self-test -v -s $SPEC -I $FAILING -b localhost:$QPID_PORT $PYTHON_TESTS || exit 1
+}
+
 if test -d ../../../python ;  then
     cd ../../../python
-    test -z "$QPID_NO_PREVIEW" && ./run-tests --skip-self-test -v -s ../specs/amqp.0-10-preview.xml -I cpp_failing_0-10_preview.txt -b localhost:$QPID_PORT $PYTHON_TESTS
-    ./run-tests --skip-self-test -v -s "0-10" -I cpp_failing_0-10.txt -b localhost:$QPID_PORT $PYTHON_TESTS
+    run 0-10 cpp_failing_0-10.txt
+    test -z "$QPID_NO_PREVIEW" && run ../specs/amqp.0-10-preview.xml cpp_failing_0-10_preview.txt
 else
     echo Warning: python tests not found.
 fi