1from string import StringBuilder 

2 

3class FifoError(Error): 

4 message: string 

5 

6@generic(T) 

7class Fifo: 

8 """A fixed size fifo. 

9 

10 """ 

11 

12 _read_index: i64 

13 _read_length: i64 

14 _items: [T] 

15 

16 func __init__(self, capacity: i64): 

17 self._read_index = 0 

18 self._read_length = 0 

19 self._items = [] 

20 

21 for _ in range(capacity): 

22 self._items.append(default(T)) 

23 

24 func length(self) -> i64: 

25 """Returns the number of items in the fifo. 

26 

27 """ 

28 

29 return self._read_length 

30 

31 func is_full(self) -> bool: 

32 """Returns True if the fifo is full, False otherwise. 

33 

34 """ 

35 

36 return self._read_length == self._items.length() 

37 

38 func is_empty(self) -> bool: 

39 """Returns True if the fifo is empty, False otherwise. 

40 

41 """ 

42 

43 return self._read_length == 0 

44 

45 func push(self, item: T): 

46 """Push an item on the fifo. 

47 

48 """ 

49 

50 if self.is_full(): 

51 raise FifoError("cannot push to full fifo") 

52 

53 write_index = (self._read_index + self._read_length) % self._items.length() 

54 self._items[write_index] = item 

55 self._read_length += 1 

56 

57 func pop(self) -> T: 

58 """Pop an item from the fifo. 

59 

60 """ 

61 

62 if self.is_empty(): 

63 raise FifoError("cannot pop from empty fifo") 

64 

65 item = self._items[self._read_index] 

66 self._read_index += 1 

67 self._read_index %= self._items.length() 

68 self._read_length -= 1 

69 

70 return item 

71 

72 func __str__(self) -> string: 

73 builder = StringBuilder() 

74 builder += "Fifo(items=[" 

75 

76 for offset in range(i64(self._read_length)): 

77 index = self._read_index + offset 

78 index %= self._items.length() 

79 

80 if offset != 0: 

81 builder += ", " 

82 

83 builder += str(self._items[index]) 

84 

85 builder += "])" 

86 

87 return builder.to_string() 

88 

89test fifo(): 

90 fifo = Fifo[string](10) 

91 assert fifo.length() == 0 

92 assert fifo.is_empty() 

93 assert not fifo.is_full() 

94 

95 try: 

96 message = "" 

97 fifo.pop() 

98 except FifoError as error: 

99 message = error.message 

100 

101 assert message == "cannot pop from empty fifo" 

102 

103 fifo.push("foo") 

104 assert fifo.pop() == "foo" 

105 

106 try: 

107 message = "" 

108 fifo.pop() 

109 except FifoError as error: 

110 message = error.message 

111 

112 assert message == "cannot pop from empty fifo" 

113 

114 for i in range(10): 

115 assert not fifo.is_full() 

116 fifo.push(str(i)) 

117 assert fifo.length() == i + 1 

118 assert not fifo.is_empty() 

119 

120 try: 

121 message = "" 

122 fifo.push("11") 

123 except FifoError as error: 

124 message = error.message 

125 

126 assert message == "cannot push to full fifo" 

127 

128 assert fifo.length() == 10 

129 assert fifo.is_full() 

130 assert not fifo.is_empty() 

131 

132 for i in range(10): 

133 assert fifo.pop() == str(i) 

134 assert fifo.length() == 9 - i 

135 

136test str(): 

137 fifo = Fifo[string](10) 

138 assert str(fifo) == "Fifo(items=[])" 

139 

140 fifo.push("1") 

141 # ToDo: Support optional. 

142 # fifo.push(None) 

143 fifo.push("2") 

144 # assert str(fifo) == "Fifo(items=[\"1\", None, \"2\"])" 

145 assert str(fifo) == "Fifo(items=[\"1\", \"2\"])" 

146 

147 # fifo.pop() 

148 fifo.pop() 

149 assert str(fifo) == "Fifo(items=[\"2\"])" 

150 

151test capacity_zero(): 

152 fifo = Fifo[bytes](0) 

153 assert fifo.is_empty() 

154 assert fifo.is_full() 

155 

156test i32(): 

157 fifo = Fifo[i32](1) 

158 fifo.push(5) 

159 assert fifo.pop() == 5